# Enhanced gcode file management and analysis
#
# Copyright (C) 2020 Eric Callahan <arksine.code@gmail.com>
#
# This file may be distributed under the terms of the GNU GPLv3 license.
from __future__ import annotations
import os
import sys
import pathlib
import shutil
import logging
import json
import tempfile
import asyncio
import zipfile
import time
from copy import deepcopy
from inotify_simple import INotify
from inotify_simple import flags as iFlags
from ...utils import source_info
# Annotation imports
from typing import (
TYPE_CHECKING,
Any,
Tuple,
Optional,
Union,
Dict,
List,
Set,
Coroutine,
Awaitable,
Callable,
TypeVar,
Type,
cast,
)
if TYPE_CHECKING:
from inotify_simple import Event as InotifyEvent
from ...confighelper import ConfigHelper
from ...common import WebRequest
from ...klippy_connection import KlippyConnection
from .. import database
from .. import klippy_apis
from .. import shell_command
from ..job_queue import JobQueue
from ..job_state import JobState
StrOrPath = Union[str, pathlib.Path]
DBComp = database.MoonrakerDatabase
APIComp = klippy_apis.KlippyAPI
SCMDComp = shell_command.ShellCommandFactory
_T = TypeVar("_T")
VALID_GCODE_EXTS = ['.gcode', '.g', '.gco', '.ufp', '.nc']
METADATA_SCRIPT = os.path.abspath(os.path.join(
os.path.dirname(__file__), "metadata.py"))
WATCH_FLAGS = iFlags.CREATE | iFlags.DELETE | iFlags.MODIFY \
| iFlags.MOVED_TO | iFlags.MOVED_FROM | iFlags.ONLYDIR \
| iFlags.CLOSE_WRITE
class FileManager:
def __init__(self, config: ConfigHelper) -> None:
self.server = config.get_server()
self.event_loop = self.server.get_event_loop()
self.reserved_paths: Dict[str, Tuple[pathlib.Path, bool]] = {}
self.full_access_roots: Set[str] = set()
self.file_paths: Dict[str, str] = {}
app_args = self.server.get_app_args()
self.datapath = pathlib.Path(app_args["data_path"])
srcdir = str(source_info.source_path())
self.add_reserved_path("moonraker", srcdir, False)
db: DBComp = self.server.load_component(config, "database")
db_path = db.get_database_path()
self.add_reserved_path("database", db_path, False)
self.add_reserved_path("certs", self.datapath.joinpath("certs"), False)
self.add_reserved_path(
"systemd", self.datapath.joinpath("systemd"), False
)
self.gcode_metadata = MetadataStorage(config, db)
self.sync_lock = NotifySyncLock(config)
avail_observers: Dict[str, Type[BaseFileSystemObserver]] = {
"none": BaseFileSystemObserver,
"inotify": InotifyObserver
}
observer = config.get("file_system_observer", "inotify").lower()
obs_class = avail_observers.get(observer)
if obs_class is None:
self.server.add_warning(
f"[file_manager]: Invalid value '{observer}' for option "
"'file_system_observer'. Falling back to no observer."
)
obs_class = BaseFileSystemObserver
if observer == "none":
logging.info("File System Observation is disabled")
else:
logging.info(f"Using File System Observer: {observer}")
self.fs_observer = obs_class(
config, self, self.gcode_metadata, self.sync_lock
)
self.scheduled_notifications: Dict[str, asyncio.TimerHandle] = {}
self.fixed_path_args: Dict[str, Any] = {}
self.queue_gcodes: bool = config.getboolean('queue_gcode_uploads', False)
# Register file management endpoints
self.server.register_endpoint(
"/server/files/list", ['GET'], self._handle_filelist_request)
self.server.register_endpoint(
"/server/files/metadata", ['GET'], self._handle_metadata_request)
self.server.register_endpoint(
"/server/files/metascan", ['POST'], self._handle_metascan_request)
self.server.register_endpoint(
"/server/files/thumbnails", ['GET'], self._handle_list_thumbs)
self.server.register_endpoint(
"/server/files/roots", ['GET'], self._handle_list_roots)
self.server.register_endpoint(
"/server/files/directory", ['GET', 'POST', 'DELETE'],
self._handle_directory_request)
self.server.register_endpoint(
"/server/files/move", ['POST'], self._handle_file_move_copy)
self.server.register_endpoint(
"/server/files/copy", ['POST'], self._handle_file_move_copy)
self.server.register_endpoint(
"/server/files/zip", ['POST'], self._handle_zip_files)
self.server.register_endpoint(
"/server/files/delete_file", ['DELETE'], self._handle_file_delete,
transports=["websocket"])
# register client notificaitons
self.server.register_notification("file_manager:filelist_changed")
self.server.register_event_handler(
"server:klippy_identified", self._update_fixed_paths)
# Register Data Folders
config.get('config_path', None, deprecate=True)
self.register_data_folder("config", full_access=True)
config.get('log_path', None, deprecate=True)
self.register_data_folder("logs")
gc_path = self.register_data_folder("gcodes", full_access=True)
if gc_path.is_dir():
prune: bool = True
saved_gc_dir: str = db.get_item(
"moonraker", "file_manager.gcode_path", ""
).result()
is_empty = next(gc_path.iterdir(), None) is None
if is_empty and saved_gc_dir:
saved_path = pathlib.Path(saved_gc_dir)
if (
saved_path.is_dir() and
next(saved_path.iterdir(), None) is not None
):
logging.info(
f"Legacy GCode Path found at '{saved_path}', "
"aborting metadata prune"
)
prune = False
if prune:
self.gcode_metadata.prune_storage()
async def component_init(self):
self.fs_observer.initialize()
def _update_fixed_paths(self) -> None:
kinfo = self.server.get_klippy_info()
paths: Dict[str, Any] = \
{k: kinfo.get(k) for k in
['klipper_path', 'python_path',
'log_file', 'config_file']}
if paths == self.fixed_path_args:
# No change in fixed paths
return
self.fixed_path_args = paths
str_paths = "\n".join([f"{k}: {v}" for k, v in paths.items()])
logging.debug(f"\nUpdating Fixed Paths:\n{str_paths}")
# Register path for example configs
klipper_path = paths.get('klipper_path', None)
if klipper_path is not None:
self.reserved_paths.pop("klipper", None)
self.add_reserved_path("klipper", klipper_path)
example_cfg_path = os.path.join(klipper_path, "config")
self.register_directory("config_examples", example_cfg_path)
docs_path = os.path.join(klipper_path, "docs")
self.register_directory("docs", docs_path)
# Register log path
log_file = paths.get('log_file')
if log_file is not None:
log_path: str = os.path.abspath(os.path.expanduser(log_file))
self.server.register_static_file_handler(
"klippy.log", log_path, force=True)
# Validate config file
cfg_file: Optional[str] = paths.get("config_file")
cfg_parent = self.file_paths.get("config")
if cfg_file is not None and cfg_parent is not None:
cfg_path = pathlib.Path(cfg_file).expanduser()
par_path = pathlib.Path(cfg_parent)
if (
par_path in cfg_path.parents or
par_path.resolve() in cfg_path.resolve().parents
):
self.server.remove_warning("klipper_config")
else:
self.server.add_warning(
"file_manager: Klipper configuration file not located in "
"'config' folder.\n\n"
f"Klipper Config Path: {cfg_path}\n\n"
f"Config Folder: {par_path}",
warn_id="klipper_config"
)
def validate_gcode_path(self, gc_path: str) -> None:
gc_dir = pathlib.Path(gc_path).expanduser()
if "gcodes" in self.file_paths:
expected = self.file_paths["gcodes"]
if not gc_dir.exists() or not gc_dir.samefile(expected):
self.server.add_warning(
"GCode path received from Klipper does not match expected "
"location.\n\n"
f"Received: '{gc_dir}'\nExpected: '{expected}'\n\n"
"Modify the [virtual_sdcard] section Klipper's "
"configuration to correct this error.\n\n"
f"[virtual_sdcard]\npath: {expected}",
warn_id="gcode_path"
)
else:
self.server.remove_warning("gcode_path")
def register_data_folder(
self, folder_name: str, full_access: bool = False
) -> pathlib.Path:
new_path = self.datapath.joinpath(folder_name)
if not new_path.exists():
try:
new_path.mkdir()
except Exception:
pass
self.register_directory(folder_name, str(new_path), full_access)
return new_path
def disable_write_access(self):
self.full_access_roots.clear()
def check_write_enabled(self):
if not self.full_access_roots:
raise self.server.error(
"Write access is currently disabled. Check notifications "
"for warnings."
)
def register_directory(self,
root: str,
path: Optional[str],
full_access: bool = False
) -> bool:
if path is None:
return False
path = os.path.abspath(os.path.expanduser(path))
if os.path.islink(path):
path = os.path.realpath(path)
if not os.path.isdir(path) or path == "/":
self.server.add_warning(
f"Supplied path ({path}) for ({root}) is invalid. Make sure\n"
"that the path exists and is not the file system root.")
return False
permissions = os.R_OK
if full_access:
permissions |= os.W_OK
self.full_access_roots.add(root)
if not os.access(path, permissions):
self.server.add_warning(
f"Moonraker does not have permission to access path "
f"({path}) for ({root}).")
return False
if path != self.file_paths.get(root, ""):
self.file_paths[root] = path
self.server.register_static_file_handler(root, path)
if root == "gcodes":
# scan for metadata changes
self.gcode_metadata.update_gcode_path(path)
if full_access:
# Refresh the file list and add watches
self.fs_observer.add_root_watch(root, path)
elif self.server.is_running():
self._sched_changed_event("root_update", root, path, immediate=True)
return True
def check_reserved_path(
self,
req_path: StrOrPath,
need_write: bool,
raise_error: bool = True
) -> bool:
if isinstance(req_path, str):
req_path = pathlib.Path(req_path)
req_path = req_path.expanduser().resolve()
if ".git" in req_path.parts:
if raise_error:
raise self.server.error(
"Access to .git folders is forbidden", 403
)
return True
for name, (res_path, can_read) in self.reserved_paths.items():
if (
(res_path == req_path or res_path in req_path.parents) and
(need_write or not can_read)
):
if not raise_error:
return True
raise self.server.error(
f"Access to file {req_path.name} forbidden by reserved "
f"path '{name}'", 403
)
return False
def add_reserved_path(
self, name: str, res_path: StrOrPath, read_access: bool = True
) -> bool:
if name in self.reserved_paths:
return False
if isinstance(res_path, str):
res_path = pathlib.Path(res_path)
res_path = res_path.expanduser().resolve()
self.reserved_paths[name] = (res_path, read_access)
return True
def get_directory(self, root: str = "gcodes") -> str:
return self.file_paths.get(root, "")
def get_registered_dirs(self) -> List[str]:
return list(self.file_paths.keys())
def get_fixed_path_args(self) -> Dict[str, Any]:
return dict(self.fixed_path_args)
def get_relative_path(self, root: str, full_path: str) -> str:
root_dir = self.file_paths.get(root, None)
if root_dir is None or not full_path.startswith(root_dir):
return ""
return os.path.relpath(full_path, start=root_dir)
def get_metadata_storage(self) -> MetadataStorage:
return self.gcode_metadata
def check_file_exists(self, root: str, filename: str) -> bool:
root_dir = self.file_paths.get(root, "")
file_path = os.path.join(root_dir, filename)
return os.path.exists(file_path)
def can_access_path(self, path: StrOrPath) -> bool:
if isinstance(path, str):
path = pathlib.Path(path)
path = path.expanduser().resolve()
for registered in self.file_paths.values():
reg_root_path = pathlib.Path(registered).resolve()
if reg_root_path in path.parents:
return not self.check_reserved_path(path, False, False)
return False
def upload_queue_enabled(self) -> bool:
return self.queue_gcodes
async def _handle_filelist_request(self,
web_request: WebRequest
) -> List[Dict[str, Any]]:
root = web_request.get_str('root', "gcodes")
flist = self.get_file_list(root, list_format=True)
return cast(List[Dict[str, Any]], flist)
async def _handle_metadata_request(self,
web_request: WebRequest
) -> Dict[str, Any]:
requested_file: str = web_request.get_str('filename')
metadata: Optional[Dict[str, Any]]
metadata = self.gcode_metadata.get(requested_file, None)
if metadata is None:
raise self.server.error(
f"Metadata not available for <{requested_file}>", 404)
metadata['filename'] = requested_file
return metadata
async def _handle_metascan_request(
self, web_request: WebRequest
) -> Dict[str, Any]:
async with self.sync_lock:
requested_file: str = web_request.get_str('filename')
gcpath = pathlib.Path(self.file_paths["gcodes"]).joinpath(requested_file)
if not gcpath.is_file():
raise self.server.error(f"File '{requested_file}' does not exist", 404)
if gcpath.suffix not in VALID_GCODE_EXTS:
raise self.server.error(f"File {gcpath} is not a valid gcode file")
# remove metadata and force a rescan
ret = self.gcode_metadata.remove_file_metadata(requested_file)
if ret is not None:
await ret
path_info = self.get_path_info(gcpath, "gcodes")
evt = self.gcode_metadata.parse_metadata(requested_file, path_info)
await evt.wait()
metadata: Optional[Dict[str, Any]]
metadata = self.gcode_metadata.get(requested_file, None)
if metadata is None:
raise self.server.error(
f"Failed to parse metadata for file '{requested_file}'", 500)
metadata['filename'] = requested_file
return metadata
async def _handle_list_roots(
self, web_request: WebRequest
) -> List[Dict[str, Any]]:
root_list: List[Dict[str, Any]] = []
for name, path in self.file_paths.items():
perms = "rw" if name in self.full_access_roots else "r"
root_list.append({
"name": name,
"path": path,
"permissions": perms
})
return root_list
async def _handle_list_thumbs(
self, web_request: WebRequest
) -> List[Dict[str, Any]]:
requested_file: str = web_request.get_str("filename")
metadata: Optional[Dict[str, Any]]
metadata = self.gcode_metadata.get(requested_file, None)
if metadata is None:
return []
if "thumbnails" not in metadata:
return []
thumblist: List[Dict[str, Any]] = metadata["thumbnails"]
for info in thumblist:
relpath: Optional[str] = info.pop("relative_path", None)
if relpath is None:
continue
thumbpath = pathlib.Path(requested_file).parent.joinpath(relpath)
info["thumbnail_path"] = str(thumbpath)
return thumblist
async def _handle_directory_request(self,
web_request: WebRequest
) -> Dict[str, Any]:
directory = web_request.get_str('path', "gcodes")
root, dir_path = self._convert_request_path(directory)
method = web_request.get_action()
if method == 'GET':
is_extended = web_request.get_boolean('extended', False)
# Get list of files and subdirectories for this target
dir_info = self._list_directory(dir_path, root, is_extended)
return dir_info
async with self.sync_lock:
self.check_reserved_path(dir_path, True)
action = "create_dir"
if method == 'POST' and root in self.full_access_roots:
# Create a new directory
self.sync_lock.setup("create_dir", dir_path)
try:
os.mkdir(dir_path)
except Exception as e:
raise self.server.error(str(e))
self.fs_observer.on_item_create(root, dir_path, is_dir=True)
elif method == 'DELETE' and root in self.full_access_roots:
# Remove a directory
action = "delete_dir"
if directory.strip("/") == root:
raise self.server.error(
"Cannot delete root directory")
if not os.path.isdir(dir_path):
raise self.server.error(
f"Directory does not exist ({directory})")
self.sync_lock.setup("delete_dir", dir_path)
force = web_request.get_boolean('force', False)
if force:
# Make sure that the directory does not contain a file
# loaded by the virtual_sdcard
self._handle_operation_check(dir_path)
try:
await self.event_loop.run_in_thread(
shutil.rmtree, dir_path)
except Exception:
raise
else:
try:
os.rmdir(dir_path)
except Exception as e:
raise self.server.error(str(e))
self.fs_observer.on_item_delete(root, dir_path, is_dir=True)
else:
raise self.server.error("Operation Not Supported", 405)
return self._sched_changed_event(action, root, dir_path)
def _handle_operation_check(self, requested_path: str) -> bool:
if not self.get_relative_path("gcodes", requested_path):
# Path not in the gcodes path
return True
kconn: KlippyConnection
kconn = self.server.lookup_component("klippy_connection")
job_state: JobState = self.server.lookup_component("job_state")
last_stats = job_state.get_last_stats()
loaded_file: str = last_stats.get('filename', "")
state: str = last_stats.get('state', "")
gc_path = self.file_paths.get('gcodes', "")
full_path = os.path.join(gc_path, loaded_file)
is_printing = kconn.is_ready() and state in ["printing", "paused"]
if loaded_file and is_printing:
if os.path.isdir(requested_path):
# Check to see of the loaded file is in the request
if full_path.startswith(requested_path):
raise self.server.error("File currently in use", 403)
elif full_path == requested_path:
raise self.server.error("File currently in use", 403)
return not is_printing
def _convert_request_path(self, request_path: str) -> Tuple[str, str]:
# Parse the root, relative path, and disk path from a remote request
parts = os.path.normpath(request_path).strip("/").split("/", 1)
if not parts:
raise self.server.error(f"Invalid path: {request_path}")
root = parts[0]
if root not in self.file_paths:
raise self.server.error(f"Invalid root path ({root})")
root_path = dest_path = self.file_paths[root]
if len(parts) > 1:
dest_path = os.path.abspath(os.path.join(dest_path, parts[1]))
if not dest_path.startswith(root_path):
raise self.server.error(
f"Invalid path request, '{request_path}'' is outside "
f"root '{root}'")
return root, dest_path
async def _handle_file_move_copy(self,
web_request: WebRequest
) -> Dict[str, Any]:
source: str = web_request.get_str("source")
destination: str = web_request.get_str("dest")
ep = web_request.get_endpoint()
source_root, source_path = self._convert_request_path(source)
dest_root, dest_path = self._convert_request_path(destination)
if dest_root not in self.full_access_roots:
raise self.server.error(
f"Destination path is read-only: {dest_root}")
self.check_reserved_path(source_path, False)
self.check_reserved_path(dest_path, True)
async with self.sync_lock:
if not os.path.exists(source_path):
raise self.server.error(f"File {source_path} does not exist")
# make sure the destination is not in use
if os.path.exists(dest_path):
self._handle_operation_check(dest_path)
src_info: Tuple[Optional[str], ...] = (None, None)
if ep == "/server/files/move":
if source_root not in self.full_access_roots:
raise self.server.error(
f"Source path is read-only, cannot move: {source_root}")
# if moving the file, make sure the source is not in use
self._handle_operation_check(source_path)
op_func: Callable[..., str] = shutil.move
action = "move_dir" if os.path.isdir(source_path) else "move_file"
src_info = (source_root, source_path)
elif ep == "/server/files/copy":
if os.path.isdir(source_path):
action = "create_dir"
op_func = shutil.copytree
else:
action = "create_file"
source_base = os.path.basename(source_path)
if (
os.path.isfile(dest_path) or
os.path.isfile(os.path.join(dest_path, source_base))
):
action = "modify_file"
op_func = shutil.copy2
self.sync_lock.setup(action, dest_path, move_copy=True)
try:
full_dest = await self.event_loop.run_in_thread(
op_func, source_path, dest_path)
if dest_root == "gcodes" and self.fs_observer.has_fast_observe:
await self.sync_lock.wait_inotify_event(full_dest)
except Exception as e:
raise self.server.error(str(e)) from e
if action.startswith("move"):
ret = self.fs_observer.on_item_move(
source_root, dest_root, source_path, full_dest
)
else:
ret = self.fs_observer.on_item_copy(dest_root, full_dest)
if ret is not None:
await ret
return self._sched_changed_event(
action, dest_root, full_dest, src_info[0], src_info[1]
)
async def _handle_zip_files(
self, web_request: WebRequest
) -> Dict[str, Any]:
async with self.sync_lock:
store_only = web_request.get_boolean("store_only", False)
suffix = time.strftime("%Y%m%d-%H%M%S", time.localtime())
dest: str = web_request.get_str(
"dest", f"config/collection-{suffix}.zip"
)
dest_root, dest_str_path = self._convert_request_path(dest)
if dest_root not in self.full_access_roots:
raise self.server.error(
f"Destination Root '{dest_root}' is read-only"
)
dest_path = pathlib.Path(dest_str_path)
self.check_reserved_path(dest_path, True)
if dest_path.is_dir():
raise self.server.error(
f"Cannot create archive at '{dest_path}'. Path exists "
"as a directory."
)
elif not dest_path.parent.exists():
raise self.server.error(
f"Cannot create archive at '{dest_path}'. Parent "
"directory does not exist."
)
items = web_request.get_list("items")
if not items:
raise self.server.error(
"At least one file or directory must be specified"
)
self.sync_lock.setup("create_file", dest_path)
await self.event_loop.run_in_thread(
self._zip_files, items, dest_path, store_only
)
self.fs_observer.on_item_create(dest_root, dest_path)
ret = self._sched_changed_event("create_file", dest_root, str(dest_path))
return {
"destination": ret["item"],
"action": "zip_files"
}
def _zip_files(
self,
item_list: List[str],
destination: StrOrPath,
store_only: bool = False
) -> None:
if isinstance(destination, str):
destination = pathlib.Path(destination).expanduser().resolve()
tmpdir = pathlib.Path(tempfile.gettempdir())
temp_dest = tmpdir.joinpath(destination.name)
processed: Set[Tuple[int, int]] = set()
cptype = zipfile.ZIP_STORED if store_only else zipfile.ZIP_DEFLATED
with zipfile.ZipFile(str(temp_dest), "w", compression=cptype) as zf:
for item in item_list:
root, str_path = self._convert_request_path(item)
root_path = pathlib.Path(self.file_paths[root])
item_path = pathlib.Path(str_path)
self.check_reserved_path(item_path, False)
if not item_path.exists():
raise self.server.error(
f"No file/directory exits at '{item}'"
)
if item_path.is_file():
st = item_path.stat()
ident = (st.st_dev, st.st_ino)
if ident in processed:
continue
processed.add(ident)
rel_path = item_path.relative_to(root_path.parent)
zf.write(str(item_path), arcname=str(rel_path))
continue
elif not item_path.is_dir():
raise self.server.error(
f"Item at path '{item}' is not a valid file or "
"directory"
)
for child_path in item_path.iterdir():
if child_path.is_file():
if self.check_reserved_path(child_path, False, False):
continue
st = child_path.stat()
ident = (st.st_dev, st.st_ino)
if ident in processed:
continue
processed.add(ident)
rel_path = child_path.relative_to(root_path.parent)
try:
zf.write(str(child_path), arcname=str(rel_path))
except PermissionError:
continue
shutil.move(str(temp_dest), str(destination))
def _list_directory(self,
path: str,
root: str,
is_extended: bool = False
) -> Dict[str, Any]:
if not os.path.isdir(path):
raise self.server.error(
f"Directory does not exist ({path})")
self.check_reserved_path(path, False)
flist: Dict[str, Any] = {'dirs': [], 'files': []}
for fname in os.listdir(path):
full_path = os.path.join(path, fname)
if not os.path.exists(full_path):
continue
path_info = self.get_path_info(full_path, root)
if os.path.isdir(full_path):
path_info['dirname'] = fname
flist['dirs'].append(path_info)
elif os.path.isfile(full_path):
path_info['filename'] = fname
# Check to see if a filelist update is necessary
ext = os.path.splitext(fname)[-1].lower()
if (
root == "gcodes" and
ext in VALID_GCODE_EXTS and
is_extended
):
rel_path = self.get_relative_path(root, full_path)
metadata: Dict[str, Any] = self.gcode_metadata.get(
rel_path, {})
path_info.update(metadata)
flist['files'].append(path_info)
usage = shutil.disk_usage(path)
flist['disk_usage'] = usage._asdict()
flist['root_info'] = {
'name': root,
'permissions': "rw" if root in self.full_access_roots else "r"
}
return flist
def get_path_info(
self, path: StrOrPath, root: str, raise_error: bool = True
) -> Dict[str, Any]:
if isinstance(path, str):
path = pathlib.Path(path)
real_path = path.resolve()
try:
fstat = path.stat()
except Exception:
if raise_error:
raise
return {"modified": 0, "size": 0, "permissions": ""}
if ".git" in real_path.parts:
permissions = ""
else:
permissions = "rw"
if (
root not in self.full_access_roots or
(path.is_symlink() and path.is_file())
):
permissions = "r"
for name, (res_path, can_read) in self.reserved_paths.items():
if (res_path == real_path or res_path in real_path.parents):
if not can_read:
permissions = ""
break
permissions = "r"
return {
'modified': fstat.st_mtime,
'size': fstat.st_size,
'permissions': permissions
}
def gen_temp_upload_path(self) -> str:
loop_time = int(self.event_loop.get_loop_time())
return os.path.join(
tempfile.gettempdir(),
f"moonraker.upload-{loop_time}.mru")
async def finalize_upload(self,
form_args: Dict[str, Any]
) -> Dict[str, Any]:
# lookup root file path
async with self.sync_lock:
try:
upload_info = self._parse_upload_args(form_args)
self.check_reserved_path(upload_info["dest_path"], True)
self.sync_lock.setup("create_file", upload_info["dest_path"])
root = upload_info['root']
if root not in self.full_access_roots:
raise self.server.error(f"Invalid root request: {root}")
if root == "gcodes" and upload_info['ext'] in VALID_GCODE_EXTS:
result = await self._finish_gcode_upload(upload_info)
else:
result = await self._finish_standard_upload(upload_info)
except Exception:
try:
os.remove(form_args['tmp_file_path'])
except Exception:
pass
raise
return result
def _parse_upload_args(self,
upload_args: Dict[str, Any]
) -> Dict[str, Any]:
if 'filename' not in upload_args:
raise self.server.error(
"No file name specifed in upload form")
# check relative path
root: str = upload_args.get('root', "gcodes").lower()
if root not in self.file_paths:
raise self.server.error(f"Root {root} not available")
root_path = self.file_paths[root]
dir_path: str = upload_args.get('path', "").lstrip("/")
if os.path.isfile(root_path):
filename: str = os.path.basename(root_path)
dest_path = root_path
dir_path = ""
else:
filename = upload_args['filename'].strip().lstrip("/")
if dir_path:
filename = os.path.join(dir_path, filename)
dest_path = os.path.abspath(os.path.join(root_path, filename))
# Validate the path. Don't allow uploads to a parent of the root
if not dest_path.startswith(root_path):
raise self.server.error(
f"Cannot write to path: {dest_path}")
start_print: bool = upload_args.get('print', "false") == "true"
f_ext = os.path.splitext(dest_path)[-1].lower()
unzip_ufp = f_ext == ".ufp" and root == "gcodes"
if unzip_ufp:
filename = os.path.splitext(filename)[0] + ".gcode"
dest_path = os.path.splitext(dest_path)[0] + ".gcode"
if os.path.isfile(dest_path) and not os.access(dest_path, os.W_OK):
raise self.server.error(f"File is read-only: {dest_path}")
return {
'root': root,
'filename': filename,
'dir_path': dir_path,
'dest_path': dest_path,
'tmp_file_path': upload_args['tmp_file_path'],
'start_print': start_print,
'unzip_ufp': unzip_ufp,
'ext': f_ext,
"is_link": os.path.islink(dest_path)
}
async def _finish_gcode_upload(
self, upload_info: Dict[str, Any]
) -> Dict[str, Any]:
# Verify that the operation can be done if attempting to upload a gcode
can_start: bool = False
try:
check_path: str = upload_info['dest_path']
can_start = self._handle_operation_check(check_path)
except self.server.error as e:
if e.status_code == 403:
raise self.server.error(
"File is loaded, upload not permitted", 403)
finfo = await self._process_uploaded_file(upload_info)
await self.gcode_metadata.parse_metadata(
upload_info['filename'], finfo).wait()
started: bool = False
queued: bool = False
if upload_info['start_print']:
if can_start:
kapis: APIComp = self.server.lookup_component('klippy_apis')
try:
await kapis.start_print(upload_info['filename'])
except self.server.error:
# Attempt to start print failed
pass
else:
started = True
if self.queue_gcodes and not started:
job_queue: JobQueue = self.server.lookup_component('job_queue')
await job_queue.queue_job(
upload_info['filename'], check_exists=False)
queued = True
self.fs_observer.on_item_create("gcodes", upload_info["dest_path"])
result = dict(self._sched_changed_event(
"create_file", "gcodes", upload_info["dest_path"],
immediate=upload_info["is_link"]
))
result.update({"print_started": started, "print_queued": queued})
return result
async def _finish_standard_upload(
self, upload_info: Dict[str, Any]
) -> Dict[str, Any]:
await self._process_uploaded_file(upload_info)
dest_path: str = upload_info["dest_path"]
root: str = upload_info["root"]
self.fs_observer.on_item_create(root, dest_path)
return self._sched_changed_event(
"create_file", root, dest_path, immediate=upload_info["is_link"]
)
async def _process_uploaded_file(self,
upload_info: Dict[str, Any]
) -> Dict[str, Any]:
try:
if upload_info['dir_path']:
cur_path = self.file_paths[upload_info['root']]
dirs: List[str]
dirs = upload_info['dir_path'].strip('/').split('/')
for subdir in dirs:
cur_path = os.path.join(cur_path, subdir)
if os.path.exists(cur_path):
continue
os.mkdir(cur_path)
# wait for inotify to create a watch before proceeding
await asyncio.sleep(.1)
if upload_info['unzip_ufp']:
tmp_path = upload_info['tmp_file_path']
finfo = self.get_path_info(tmp_path, upload_info['root'])
finfo['ufp_path'] = tmp_path
else:
dest_path = upload_info['dest_path']
if upload_info["is_link"]:
dest_path = os.path.realpath(dest_path)
shutil.move(
upload_info['tmp_file_path'], dest_path)
finfo = self.get_path_info(upload_info['dest_path'],
upload_info['root'])
except Exception:
logging.exception("Upload Write Error")
raise self.server.error("Unable to save file", 500)
return finfo
def get_file_list(self,
root: str,
list_format: bool = False
) -> Union[Dict[str, Any], List[Dict[str, Any]]]:
# Use os.walk find files in sd path and subdirs
filelist: Dict[str, Any] = {}
path = self.file_paths.get(root, None)
if path is None or not os.path.isdir(path):
msg = f"Failed to build file list, invalid path: {root}: {path}"
logging.info(msg)
raise self.server.error(msg)
logging.info(f"Updating File List <{root}>...")
st = os.stat(path)
visited_dirs = {(st.st_dev, st.st_ino)}
for dir_path, dir_names, files in os.walk(path, followlinks=True):
scan_dirs: List[str] = []
# Filter out directories that have already been visted. This
# prevents infinite recrusion "followlinks" is set to True
for dname in dir_names:
full_path = os.path.join(dir_path, dname)
if not os.path.exists(full_path):
continue
st = os.stat(full_path)
key = (st.st_dev, st.st_ino)
if key not in visited_dirs:
visited_dirs.add(key)
if not self.check_reserved_path(full_path, False, False):
scan_dirs.append(dname)
dir_names[:] = scan_dirs
for name in files:
ext = os.path.splitext(name)[-1].lower()
if root == 'gcodes' and ext not in VALID_GCODE_EXTS:
continue
full_path = os.path.join(dir_path, name)
if not os.path.exists(full_path):
continue
fname = full_path[len(path) + 1:]
finfo = self.get_path_info(full_path, root)
filelist[fname] = finfo
if list_format:
flist: List[Dict[str, Any]] = []
for fname in sorted(filelist, key=str.lower):
fdict: Dict[str, Any] = {'path': fname}
fdict.update(filelist[fname])
flist.append(fdict)
return flist
return filelist
def get_file_metadata(self, filename: str) -> Dict[str, Any]:
if filename[0] == '/':
filename = filename[1:]
# Remove "gcodes" of its added. It is valid for a request to
# include to the root or assume the root is gcodes
if filename.startswith('gcodes/'):
filename = filename[7:]
return self.gcode_metadata.get(filename, {})
def list_dir(self,
directory: str,
simple_format: bool = False
) -> Union[Dict[str, Any], List[str]]:
# List a directory relative to its root.
if directory[0] == "/":
directory = directory[1:]
parts = directory.split("/", 1)
root = parts[0]
if root not in self.file_paths:
raise self.server.error(
f"Invalid Directory Request: {directory}")
path = self.file_paths[root]
if len(parts) == 1:
dir_path = path
else:
dir_path = os.path.join(path, parts[1])
if not os.path.isdir(dir_path):
raise self.server.error(
f"Directory does not exist ({dir_path})")
flist = self._list_directory(dir_path, root)
if simple_format:
simple_list = []
for dirobj in flist['dirs']:
simple_list.append("*" + dirobj['dirname'])
for fileobj in flist['files']:
fname = fileobj['filename']
ext = os.path.splitext(fname)[-1].lower()
if root == "gcodes" and ext in VALID_GCODE_EXTS:
simple_list.append(fname)
return simple_list
return flist
async def _handle_file_delete(self,
web_request: WebRequest
) -> Dict[str, Any]:
file_path: str = web_request.get_str("path")
return await self.delete_file(file_path)
async def delete_file(self, path: str) -> Dict[str, Any]:
async with self.sync_lock:
root, full_path = self._convert_request_path(path)
self.check_reserved_path(full_path, True)
if root not in self.full_access_roots:
raise self.server.error(
f"Path not available for DELETE: {path}", 405)
if not os.path.isfile(full_path):
raise self.server.error(f"Invalid file path: {path}")
try:
self._handle_operation_check(full_path)
except self.server.error as e:
if e.status_code == 403:
raise
self.sync_lock.setup("delete_file", full_path)
os.remove(full_path)
self.fs_observer.on_item_delete(root, full_path)
return self._sched_changed_event("delete_file", root, full_path)
def _sched_changed_event(
self,
action: str,
root: str,
full_path: str,
source_root: Optional[str] = None,
source_path: Optional[str] = None,
immediate: bool = False
) -> Dict[str, Any]:
rel_path = self.get_relative_path(root, full_path)
path_info = self.get_path_info(full_path, root, raise_error=False)
path_info.update({"path": rel_path, "root": root})
notify_info: Dict[str, Any] = {
"action": action,
"item": path_info
}
if source_path is not None and source_root is not None:
src_rel_path = self.get_relative_path(source_root, source_path)
notify_info['source_item'] = {'path': src_rel_path, 'root': source_root}
immediate |= not self.fs_observer.has_fast_observe
delay = .005 if immediate else 1.
key = f"{action}-{root}-{rel_path}"
handle = self.event_loop.delay_callback(
delay, self._do_notify, key, notify_info
)
if not immediate:
self.scheduled_notifications[key] = handle
return notify_info
def _do_notify(self, key: str, notify_info: Dict[str, Any]) -> None:
self.scheduled_notifications.pop(key, None)
self.server.send_event("file_manager:filelist_changed", notify_info)
def cancel_notification(self, key: str) -> None:
handle = self.scheduled_notifications.pop(key, None)
if handle is not None:
handle.cancel()
def close(self) -> None:
for hdl in self.scheduled_notifications.values():
hdl.cancel()
self.scheduled_notifications.clear()
self.fs_observer.close()
class NotifySyncLock(asyncio.Lock):
def __init__(self, config: ConfigHelper) -> None:
super().__init__()
self.server = config.get_server()
self.action: str = ""
self.dest_path: Optional[pathlib.Path] = None
self.check_pending = False
self.move_copy_fut: Optional[asyncio.Future] = None
self.sync_waiters: List[asyncio.Future] = []
self.pending_paths: Set[pathlib.Path] = set()
self.acquired_paths: Set[pathlib.Path] = set()
def setup(
self, action: str, path: StrOrPath, move_copy: bool = False
) -> None:
if not self.locked():
raise self.server.error(
"Cannot call setup unless the lock has been acquired"
)
# Called by a file manager request. Sets the destination path to sync
# with the file system observer (inotify).
if self.dest_path is not None:
logging.debug(
"NotifySync Error: Setup requested while a path is still pending"
)
self.finish()
if isinstance(path, str):
path = pathlib.Path(path)
self.dest_path = path
self.action = action
self.check_pending = move_copy
async def wait_inotify_event(self, current_path: StrOrPath) -> None:
# Called by a file manager move copy request to wait for metadata
# analysis to complete. We need to be careful here to avoid a deadlock
# or a long wait time when inotify isn't available.
if not self.check_pending:
return
if isinstance(current_path, str):
current_path = pathlib.Path(current_path)
self.dest_path = current_path
if current_path in self.acquired_paths:
# Notifcation has been recieved, no need to wait
return
self.move_copy_fut = self.server.get_event_loop().create_future()
mcfut = self.move_copy_fut
has_pending = current_path in self.pending_paths
timeout = 1200. if has_pending else 1.
for _ in range(5):
try:
await asyncio.wait_for(asyncio.shield(mcfut), timeout)
except asyncio.TimeoutError:
if timeout > 2.:
break
has_pending = current_path in self.pending_paths
timeout = 1200. if has_pending else 1.
else:
break
else:
logging.info(
f"Failed to receive an inotify event, dest path: {current_path}"
)
self.move_copy_fut = None
def finish(self) -> None:
# Called by a file manager request upon completion. The inotify observer
# can now emit the websocket notification
for waiter in self.sync_waiters:
if not waiter.done():
waiter.set_result((self.action, self.dest_path))
self.sync_waiters.clear()
self.dest_path = None
self.action = ""
self.pending_paths.clear()
self.acquired_paths.clear()
if self.move_copy_fut is not None and not self.move_copy_fut.done():
self.move_copy_fut.set_exception(
self.server.error("Move/Copy Interrupted by call to finish")
)
self.move_copy_fut = None
self.check_pending = False
def add_pending_path(self, action: str, pending_path: StrOrPath) -> None:
# Called by the inotify observer whenever a create or move event
# is detected. This is only necessary to track for move/copy actions,
# since we don't get the final destination until the request is complete.
if (
not self.check_pending or
self.dest_path is None or
action != self.action
):
return
if isinstance(pending_path, str):
pending_path = pathlib.Path(pending_path)
if self.dest_path in [pending_path, pending_path.parent]:
self.pending_paths.add(pending_path)
def check_in_request(
self, action: str, inotify_path: StrOrPath
) -> Optional[asyncio.Future]:
# Called by the inotify observer to check if request synchronization
# is necessary. If so, this method will return a future the inotify
# observer can await.
if self.dest_path is None:
return None
if isinstance(inotify_path, str):
inotify_path = pathlib.Path(inotify_path)
waiter: Optional[asyncio.Future] = None
if self.check_pending:
# The final path of move/copy requests aren't known until the request
# complete. It may be the destination path recieved from the request
# or it may be a child as of that path.
if self.move_copy_fut is not None:
# Request is complete, metadata analysis pending. We can explicitly
# check for a path match
if self.dest_path == inotify_path:
if not self.move_copy_fut.done():
self.move_copy_fut.set_result(None)
waiter = self.server.get_event_loop().create_future()
elif self.dest_path in [inotify_path, inotify_path.parent]:
# Request is still processing. This might be the notification for
# the request, it will be checked when the move/copy request awaits
self.acquired_paths.add(inotify_path)
waiter = self.server.get_event_loop().create_future()
elif self.dest_path == inotify_path:
waiter = self.server.get_event_loop().create_future()
if waiter is not None:
self._check_action(action, inotify_path)
self.sync_waiters.append(waiter)
return waiter
def _check_action(self, action: str, path: StrOrPath) -> bool:
# We aren't going to set a hard filter on the sync action, however
# we will log mismatches as they shouldn't occur
if action != self.action:
logging.info(
f"\nInotify action mismatch:\n"
f"Expected action: {self.action}, Inotify action: {action}\n"
f"Requested path: {self.dest_path}\n"
f"Inotify path: {path}\n"
f"Is move/copy: {self.check_pending}"
)
return False
return True
def release(self) -> None:
super().release()
self.finish()
class BaseFileSystemObserver:
def __init__(
self,
config: ConfigHelper,
file_manager: FileManager,
gcode_metadata: MetadataStorage,
sync_lock: NotifySyncLock
) -> None:
self.server = config.get_server()
self.event_loop = self.server.get_event_loop()
self.enable_warn = config.getboolean("enable_observer_warnings", True)
self.file_manager = file_manager
self.gcode_metadata = gcode_metadata
self.sync_lock = sync_lock
@property
def has_fast_observe(self) -> bool:
return False
def initialize(self) -> None:
pass
def add_root_watch(self, root: str, root_path: str) -> None:
# Just emit the notification
if self.server.is_running():
fm = self.file_manager
fm._sched_changed_event("root_update", root, root_path, immediate=True)
def try_move_metadata(
self,
prev_root: str,
new_root: str,
prev_path: str,
new_path: str,
is_dir: bool = False
) -> Union[bool, Awaitable]:
if new_root == "gcodes":
if prev_root == "gcodes":
# moved within the gcodes root, move metadata
fm = self.file_manager
gcm = self.gcode_metadata
prev_rel_path = fm.get_relative_path("gcodes", prev_path)
new_rel_path = fm.get_relative_path("gcodes", new_path)
if is_dir:
gcm.move_directory_metadata(prev_rel_path, new_rel_path)
else:
return gcm.move_file_metadata(prev_rel_path, new_rel_path)
else:
# move from a non-gcodes root to gcodes root needs a rescan
self.clear_metadata(prev_root, prev_path, is_dir)
return False
elif prev_root == "gcodes":
# moved out of the gcodes root, remove metadata
self.clear_metadata(prev_root, prev_path, is_dir)
return True
def clear_metadata(
self, root: str, path: str, is_dir: bool = False
) -> None:
if root == "gcodes":
rel_path = self.file_manager.get_relative_path(root, str(path))
if is_dir:
self.gcode_metadata.remove_directory_metadata(rel_path)
else:
self.gcode_metadata.remove_file_metadata(rel_path)
def parse_gcode_metadata(self, file_path: str) -> asyncio.Event:
rel_path = self.file_manager.get_relative_path("gcodes", file_path)
ext = os.path.splitext(rel_path)[-1].lower()
try:
path_info = self.file_manager.get_path_info(file_path, "gcodes")
except Exception:
path_info = {}
if (
ext not in VALID_GCODE_EXTS or
path_info.get('size', 0) == 0
):
evt = asyncio.Event()
evt.set()
return evt
if ext == ".ufp":
rel_path = os.path.splitext(rel_path)[0] + ".gcode"
path_info['ufp_path'] = file_path
return self.gcode_metadata.parse_metadata(rel_path, path_info)
def _scan_directory_metadata(
self, start_path: pathlib.Path
) -> Optional[Awaitable]:
# Use os.walk find files in sd path and subdirs
mevts: List[Coroutine] = []
st = start_path.stat()
visited_dirs = {(st.st_dev, st.st_ino)}
for parent, dirs, files in os.walk(start_path, followlinks=True):
scan_dirs: List[str] = []
# Filter out directories that have already been visted. This
# prevents infinite recrusion "followlinks" is set to True
parent_dir = pathlib.Path(parent)
for dname in dirs:
dir_path = parent_dir.joinpath(dname)
if not dir_path.exists():
continue
st = dir_path.stat()
key = (st.st_dev, st.st_ino)
if key not in visited_dirs:
visited_dirs.add(key)
scan_dirs.append(dname)
dirs[:] = scan_dirs
for fname in files:
file_path = parent_dir.joinpath(fname)
if (
not file_path.is_file() or
file_path.suffix not in VALID_GCODE_EXTS
):
continue
mevt = self.parse_gcode_metadata(str(file_path))
mevts.append(mevt.wait())
if mevts:
return asyncio.gather(*mevts)
return None
def on_item_copy(self, root: str, item_path: StrOrPath) -> Optional[Awaitable]:
if self.has_fast_observe:
return None
if isinstance(item_path, str):
item_path = pathlib.Path(item_path)
if root != "gcodes":
return None
if item_path.is_file() and item_path.suffix in VALID_GCODE_EXTS:
ret = self.parse_gcode_metadata(str(item_path))
return ret.wait()
elif item_path.is_dir():
return self._scan_directory_metadata(item_path)
return None
def on_item_move(
self,
src_root: str,
dest_root: str,
src_path: StrOrPath,
dest_path: StrOrPath
) -> Optional[Awaitable]:
if self.has_fast_observe:
return None
if isinstance(src_path, str):
src_path = pathlib.Path(src_path)
if isinstance(dest_path, str):
dest_path = pathlib.Path(dest_path)
is_dir = dest_path.is_dir()
ret = self.try_move_metadata(
src_root, dest_root, str(src_path), str(dest_path), is_dir
)
if not isinstance(ret, bool):
return ret
elif ret is False:
# Need metadata scan
if is_dir:
return self._scan_directory_metadata(dest_path)
elif dest_path.is_file() and dest_path.suffix in VALID_GCODE_EXTS:
mevt = self.parse_gcode_metadata(str(dest_path))
return mevt.wait()
return None
def on_item_create(
self, root: str, item_path: StrOrPath, is_dir: bool = False
) -> None:
pass
def on_item_delete(
self, root: str, item_path: StrOrPath, is_dir: bool = False
) -> None:
if self.has_fast_observe:
return
self.clear_metadata(root, str(item_path), is_dir)
def close(self) -> None:
pass
INOTIFY_BUNDLE_TIME = .25
INOTIFY_MOVE_TIME = 1.
class InotifyNode:
def __init__(self,
iobsvr: InotifyObserver,
parent: InotifyNode,
name: str
) -> None:
self.iobsvr = iobsvr
self.event_loop = iobsvr.event_loop
self.name = name
self.parent_node = parent
self.child_nodes: Dict[str, InotifyNode] = {}
self.watch_desc = self.iobsvr.add_watch(self)
self.pending_node_events: Dict[str, asyncio.Handle] = {}
self.pending_deleted_children: Set[Tuple[str, bool]] = set()
self.pending_file_events: Dict[str, str] = {}
self.queued_move_notificatons: List[List[str]] = []
self.is_processing_metadata = False
async def _finish_create_node(self) -> None:
# Finish a node's creation. All children that were created
# with this node (ie: a directory is copied) are bundled into
# this notification. We also scan the node to extract metadata
# here, as we know all files have been copied.
if "create_node" not in self.pending_node_events:
return
del self.pending_node_events['create_node']
node_path = self.get_path()
root = self.get_root()
# Scan child nodes for unwatched directories and metadata
self.is_processing_metadata = True
mevts: List[asyncio.Event] = self.scan_node()
if mevts:
mfuts = [e.wait() for e in mevts]
await asyncio.gather(*mfuts)
self.is_processing_metadata = False
self.iobsvr.log_nodes()
self.iobsvr.notify_filelist_changed(
"create_dir", root, node_path)
for args in self.queued_move_notificatons:
self.iobsvr.notify_filelist_changed(*args)
self.queued_move_notificatons.clear()
def _finish_delete_child(self) -> None:
# Items deleted in a child (node or file) are batched.
# Individual files get notifications if their parent
# node stil exists. Otherwise notififications are
# bundled into the topmost deleted parent.
if "delete_child" not in self.pending_node_events:
self.pending_deleted_children.clear()
return
del self.pending_node_events['delete_child']
node_path = self.get_path()
root = self.get_root()
for (name, is_node) in self.pending_deleted_children:
item_path = os.path.join(node_path, name)
item_type = "dir" if is_node else "file"
self.iobsvr.clear_metadata(root, item_path, is_node)
self.iobsvr.notify_filelist_changed(
f"delete_{item_type}", root, item_path)
self.pending_deleted_children.clear()
def scan_node(self,
visited_dirs: Set[Tuple[int, int]] = set()
) -> List[asyncio.Event]:
dir_path = self.get_path()
st = os.stat(dir_path)
if st in visited_dirs:
return []
metadata_events: List[asyncio.Event] = []
visited_dirs.add((st.st_dev, st.st_ino))
for fname in os.listdir(dir_path):
item_path = os.path.join(dir_path, fname)
if os.path.isdir(item_path):
fm = self.iobsvr.file_manager
if fm.check_reserved_path(item_path, True, False):
continue
new_child = self.create_child_node(fname, False)
if new_child is not None:
metadata_events.extend(new_child.scan_node(visited_dirs))
elif os.path.isfile(item_path) and self.get_root() == "gcodes":
mevt = self.iobsvr.parse_gcode_metadata(item_path)
metadata_events.append(mevt)
return metadata_events
def move_child_node(
self,
child_name: str,
new_name: str,
new_parent: InotifyNode
) -> None:
self.flush_delete()
child_node = self.pop_child_node(child_name)
if child_node is None:
logging.info(f"No child for node at path: {self.get_path()}")
return
prev_path = child_node.get_path()
prev_root = child_node.get_root()
child_node.name = new_name
new_parent.add_child_node(child_node)
new_path = child_node.get_path()
new_root = child_node.get_root()
logging.debug(f"Moving node from '{prev_path}' to '{new_path}'")
# Attempt to move metadata
move_res = self.iobsvr.try_move_metadata(
prev_root, new_root, prev_path, new_path, is_dir=True
)
if new_root == "gcodes":
async def _notify_move_dir():
if move_res is False:
# Need rescan
mevts = child_node.scan_node()
if mevts:
mfuts = [e.wait() for e in mevts]
await asyncio.gather(*mfuts)
self.iobsvr.notify_filelist_changed(
"move_dir", new_root, new_path, prev_root, prev_path
)
self.iobsvr.queue_gcode_notification(_notify_move_dir())
else:
self.iobsvr.notify_filelist_changed(
"move_dir", new_root, new_path, prev_root, prev_path
)
def schedule_file_event(self, file_name: str, evt_name: str) -> None:
if file_name in self.pending_file_events:
return
pending_node = self.search_pending_event("create_node")
if pending_node is not None:
pending_node.stop_event("create_node")
self.pending_file_events[file_name] = evt_name
def complete_file_write(self, file_name: str) -> None:
self.flush_delete()
evt_name = self.pending_file_events.pop(file_name, None)
if evt_name is None:
logging.info(f"Invalid file write event: {file_name}")
return
if self.is_processing():
logging.debug("Metadata is processing, suppressing write "
f"event: {file_name}")
return
pending_node = self.search_pending_event("create_node")
if pending_node is not None:
# if this event was generated as a result of a created parent
# node it should be ignored in favor of the parent event.
pending_node.reset_event("create_node", INOTIFY_BUNDLE_TIME)
return
file_path = os.path.join(self.get_path(), file_name)
root = self.get_root()
if root == "gcodes":
if self.iobsvr.need_create_notify(file_path):
async def _notify_file_write():
mevt = self.iobsvr.parse_gcode_metadata(file_path)
await mevt.wait()
self.iobsvr.notify_filelist_changed(evt_name, root, file_path)
self.iobsvr.clear_processing_file(file_path)
self.iobsvr.queue_gcode_notification(_notify_file_write())
else:
self.iobsvr.notify_filelist_changed(evt_name, root, file_path)
def add_child_node(self, node: InotifyNode) -> None:
self.child_nodes[node.name] = node
node.parent_node = self
def get_child_node(self, name: str) -> Optional[InotifyNode]:
return self.child_nodes.get(name, None)
def pop_child_node(self, name: str) -> Optional[InotifyNode]:
return self.child_nodes.pop(name, None)
def create_child_node(self,
name: str,
notify: bool = True
) -> Optional[InotifyNode]:
self.flush_delete()
if name in self.child_nodes:
return self.child_nodes[name]
try:
new_child = InotifyNode(self.iobsvr, self, name)
except Exception:
# This node is already watched under another root,
# bypass creation
return None
self.child_nodes[name] = new_child
if notify:
pending_node = self.search_pending_event("create_node")
if pending_node is None:
# schedule a pending create event for the child
new_child.add_event("create_node", INOTIFY_BUNDLE_TIME)
else:
pending_node.reset_event("create_node", INOTIFY_BUNDLE_TIME)
return new_child
def schedule_child_delete(self, child_name: str, is_node: bool) -> None:
if is_node:
child_node = self.child_nodes.pop(child_name, None)
if child_node is None:
return
self.iobsvr.remove_watch(
child_node.watch_desc, need_low_level_rm=False)
child_node.remove_event("delete_child")
self.pending_deleted_children.add((child_name, is_node))
self.add_event("delete_child", INOTIFY_BUNDLE_TIME)
def clear_watches(self) -> None:
for cnode in self.child_nodes.values():
# Delete all of the children's children
cnode.clear_watches()
self.iobsvr.remove_watch(self.watch_desc)
def get_path(self) -> str:
return os.path.join(self.parent_node.get_path(), self.name)
def get_root(self) -> str:
return self.parent_node.get_root()
def is_processing(self) -> bool:
if self.is_processing_metadata:
return True
return self.parent_node.is_processing()
def add_event(self, evt_name: str, timeout: float) -> None:
if evt_name in self.pending_node_events:
self.reset_event(evt_name, timeout)
return
callback = getattr(self, f"_finish_{evt_name}")
hdl = self.event_loop.delay_callback(timeout, callback)
self.pending_node_events[evt_name] = hdl
def reset_event(self, evt_name: str, timeout: float) -> None:
if evt_name in self.pending_node_events:
hdl = self.pending_node_events[evt_name]
hdl.cancel()
callback = getattr(self, f"_finish_{evt_name}")
hdl = self.event_loop.delay_callback(timeout, callback)
self.pending_node_events[evt_name] = hdl
def stop_event(self, evt_name: str) -> None:
if evt_name in self.pending_node_events:
hdl = self.pending_node_events[evt_name]
hdl.cancel()
def remove_event(self, evt_name: str) -> None:
hdl = self.pending_node_events.pop(evt_name, None)
if hdl is not None:
hdl.cancel()
def flush_delete(self):
if 'delete_child' not in self.pending_node_events:
return
hdl = self.pending_node_events['delete_child']
hdl.cancel()
self._finish_delete_child()
def clear_events(self, include_children: bool = True) -> None:
if include_children:
for child in self.child_nodes.values():
child.clear_events(include_children)
for hdl in self.pending_node_events.values():
hdl.cancel()
self.pending_node_events.clear()
self.pending_deleted_children.clear()
self.pending_file_events.clear()
def search_pending_event(self, name: str) -> Optional[InotifyNode]:
if name in self.pending_node_events:
return self
return self.parent_node.search_pending_event(name)
def find_pending_node(self) -> Optional[InotifyNode]:
if (
self.is_processing_metadata or
"create_node" in self.pending_node_events
):
return self
return self.parent_node.find_pending_node()
def queue_move_notification(self, args: List[str]) -> None:
if (
self.is_processing_metadata or
"create_node" in self.pending_node_events
):
self.queued_move_notificatons.append(args)
else:
if self.iobsvr.server.is_verbose_enabled():
path = self.get_path()
logging.debug(
f"Node {path} received a move notification queue request, "
f"however node is not pending: {args}"
)
self.iobsvr.notify_filelist_changed(*args)
class InotifyRootNode(InotifyNode):
def __init__(self,
iobsvr: InotifyObserver,
root_name: str,
root_path: str
) -> None:
self.root_name = root_name
super().__init__(iobsvr, self, root_path)
def get_path(self) -> str:
return self.name
def get_root(self) -> str:
return self.root_name
def search_pending_event(self, name) -> Optional[InotifyNode]:
if name in self.pending_node_events:
return self
return None
def is_processing(self) -> bool:
return self.is_processing_metadata
def find_pending_node(self) -> Optional[InotifyNode]:
if (
self.is_processing_metadata or
"create_node" in self.pending_node_events
):
return self
return None
class InotifyObserver(BaseFileSystemObserver):
def __init__(
self,
config: ConfigHelper,
file_manager: FileManager,
gcode_metadata: MetadataStorage,
sync_lock: NotifySyncLock
) -> None:
super().__init__(config, file_manager, gcode_metadata, sync_lock)
self.enable_warn = config.getboolean(
"enable_inotify_warnings", self.enable_warn, deprecate=True
)
self.inotify = INotify(nonblocking=True)
self.event_loop.add_reader(
self.inotify.fileno(), self._handle_inotify_read)
self.watched_roots: Dict[str, InotifyRootNode] = {}
self.watched_nodes: Dict[int, InotifyNode] = {}
self.pending_moves: Dict[
int, Tuple[InotifyNode, str, asyncio.Handle]] = {}
self.initialized: bool = False
self.processing_gcode_files: Set[str] = set()
self.pending_coroutines: List[Coroutine] = []
self._gc_notify_task: Optional[asyncio.Task] = None
@property
def has_fast_observe(self) -> bool:
return True
# Override and pass the callbacks from the request handlers. Inotify
# detects events quickly and takes any required actions
def on_item_create(
self, root: str, item_path: StrOrPath, is_dir: bool = False
) -> None:
pass
def on_item_delete(
self, root: str, item_path: StrOrPath, is_dir: bool = False
) -> None:
pass
def on_item_move(
self,
src_root: str,
dest_root: str,
src_path: StrOrPath,
dest_path: StrOrPath
) -> Optional[Awaitable]:
return None
def on_item_copy(self, root: str, item_path: StrOrPath) -> Optional[Awaitable]:
return None
def add_root_watch(self, root: str, root_path: str) -> None:
# remove all exisiting watches on root
if root in self.watched_roots:
old_root = self.watched_roots.pop(root)
old_root.clear_watches()
old_root.clear_events()
try:
root_node = InotifyRootNode(self, root, root_path)
except Exception:
return
self.watched_roots[root] = root_node
if self.initialized:
mevts = root_node.scan_node()
self.log_nodes()
self.event_loop.register_callback(
self._notify_root_updated, mevts, root, root_path)
def initialize(self) -> None:
for root, node in self.watched_roots.items():
evts = node.scan_node()
if not evts:
continue
root_path = node.get_path()
self.event_loop.register_callback(
self._notify_root_updated, evts, root, root_path)
if self.watched_roots:
self.log_nodes()
self.initialized = True
async def _notify_root_updated(self,
mevts: List[asyncio.Event],
root: str,
root_path: str
) -> None:
if mevts:
mfuts = [e.wait() for e in mevts]
await asyncio.gather(*mfuts)
cur_path = self.watched_roots[root].get_path()
if self.server.is_running() and cur_path == root_path:
self.notify_filelist_changed("root_update", root, root_path)
def add_watch(self, node: InotifyNode) -> int:
dir_path = node.get_path()
try:
watch: int = self.inotify.add_watch(dir_path, WATCH_FLAGS)
except Exception:
msg = (
f"Error adding inotify watch to root '{node.get_root()}', "
f"path: {dir_path}"
)
logging.exception(msg)
if self.enable_warn:
msg = f"file_manager: {msg}"
self.server.add_warning(msg, log=False)
raise
if watch in self.watched_nodes:
root = node.get_root()
cur_node = self.watched_nodes[watch]
existing_root = cur_node.get_root()
msg = (
f"Inotify watch already exists for path '{dir_path}' in "
f"root '{existing_root}', cannot add watch to requested root "
f"'{root}'. This indicates that the roots overlap."
)
if self.enable_warn:
msg = f"file_manager: {msg}"
self.server.add_warning(msg)
else:
logging.info(msg)
raise self.server.error("Watch already exists")
self.watched_nodes[watch] = node
return watch
def remove_watch(self,
wdesc: int,
need_low_level_rm: bool = True
) -> None:
node = self.watched_nodes.pop(wdesc, None)
if need_low_level_rm and node is not None:
try:
self.inotify.rm_watch(wdesc)
except Exception:
logging.exception(f"Error removing watch: '{node.get_path()}'")
def log_nodes(self) -> None:
if self.server.is_verbose_enabled():
debug_msg = f"Inotify Watches After Scan:"
for wdesc, node in self.watched_nodes.items():
wdir = node.get_path()
wroot = node.get_root()
debug_msg += f"\nRoot: {wroot}, Directory: {wdir}, " \
f"Watch: {wdesc}"
logging.debug(debug_msg)
def _handle_move_timeout(self, cookie: int, is_dir: bool):
if cookie not in self.pending_moves:
return
parent_node, name, hdl = self.pending_moves.pop(cookie)
item_path = os.path.join(parent_node.get_path(), name)
root = parent_node.get_root()
self.clear_metadata(root, item_path, is_dir)
action = "delete_file"
if is_dir:
# The supplied node is a child node
child_node = parent_node.pop_child_node(name)
if child_node is None:
return
child_node.clear_watches()
child_node.clear_events(include_children=True)
self.log_nodes()
action = "delete_dir"
self.notify_filelist_changed(action, root, item_path)
def _schedule_pending_move(self,
evt: InotifyEvent,
parent_node: InotifyNode,
is_dir: bool
) -> None:
hdl = self.event_loop.delay_callback(
INOTIFY_MOVE_TIME, self._handle_move_timeout,
evt.cookie, is_dir)
self.pending_moves[evt.cookie] = (parent_node, evt.name, hdl)
def _handle_inotify_read(self) -> None:
evt: InotifyEvent
for evt in self.inotify.read(timeout=0):
if evt.mask & iFlags.IGNORED:
continue
if evt.wd not in self.watched_nodes:
flags = " ".join([str(f) for f in iFlags.from_mask(evt.mask)])
logging.info(
f"Error, inotify watch descriptor {evt.wd} "
f"not currently tracked: name: {evt.name}, "
f"flags: {flags}")
continue
node = self.watched_nodes[evt.wd]
if evt.mask & iFlags.ISDIR:
self._process_dir_event(evt, node)
else:
self._process_file_event(evt, node)
def _process_dir_event(self, evt: InotifyEvent, node: InotifyNode) -> None:
if evt.name in ['.', ".."]:
# ignore events for self and parent
return
root = node.get_root()
node_path = node.get_path()
full_path = os.path.join(node_path, evt.name)
if evt.mask & iFlags.CREATE:
self.sync_lock.add_pending_path("create_dir", full_path)
logging.debug(f"Inotify directory create: {root}, "
f"{node_path}, {evt.name}")
node.create_child_node(evt.name)
elif evt.mask & iFlags.DELETE:
logging.debug(f"Inotify directory delete: {root}, "
f"{node_path}, {evt.name}")
node.schedule_child_delete(evt.name, True)
elif evt.mask & iFlags.MOVED_FROM:
logging.debug(f"Inotify directory move from: {root}, "
f"{node_path}, {evt.name}")
self._schedule_pending_move(evt, node, True)
elif evt.mask & iFlags.MOVED_TO:
logging.debug(f"Inotify directory move to: {root}, "
f"{node_path}, {evt.name}")
moved_evt = self.pending_moves.pop(evt.cookie, None)
if moved_evt is not None:
self.sync_lock.add_pending_path("move_dir", full_path)
# Moved from a currently watched directory
prev_parent, child_name, hdl = moved_evt
hdl.cancel()
prev_parent.move_child_node(child_name, evt.name, node)
else:
# Moved from an unwatched directory, for our
# purposes this is the same as creating a
# directory
self.sync_lock.add_pending_path("create_dir", full_path)
node.create_child_node(evt.name)
def _process_file_event(self, evt: InotifyEvent, node: InotifyNode) -> None:
ext: str = os.path.splitext(evt.name)[-1].lower()
root = node.get_root()
node_path = node.get_path()
file_path = os.path.join(node_path, evt.name)
if evt.mask & iFlags.CREATE:
logging.debug(f"Inotify file create: {root}, "
f"{node_path}, {evt.name}")
self.sync_lock.add_pending_path("create_file", file_path)
node.schedule_file_event(evt.name, "create_file")
if os.path.islink(file_path):
logging.debug(f"Inotify symlink create: {file_path}")
node.complete_file_write(evt.name)
elif evt.mask & iFlags.DELETE:
logging.debug(f"Inotify file delete: {root}, "
f"{node_path}, {evt.name}")
if root == "gcodes" and ext == ".ufp":
# Don't notify deleted ufp files
return
node.schedule_child_delete(evt.name, False)
elif evt.mask & iFlags.MOVED_FROM:
logging.debug(f"Inotify file move from: {root}, "
f"{node_path}, {evt.name}")
self._schedule_pending_move(evt, node, False)
elif evt.mask & iFlags.MOVED_TO:
logging.debug(f"Inotify file move to: {root}, "
f"{node_path}, {evt.name}")
node.flush_delete()
moved_evt = self.pending_moves.pop(evt.cookie, None)
pending_node = node.find_pending_node()
if moved_evt is not None:
# Moved from a currently watched directory
self.sync_lock.add_pending_path("move_file", file_path)
prev_parent, prev_name, hdl = moved_evt
hdl.cancel()
prev_root = prev_parent.get_root()
prev_path = os.path.join(prev_parent.get_path(), prev_name)
move_res = self.try_move_metadata(prev_root, root, prev_path, file_path)
if root == "gcodes":
coro = self._finish_gcode_move(
root, prev_root, file_path, prev_path, pending_node, move_res
)
self.queue_gcode_notification(coro)
else:
args = ["move_file", root, file_path, prev_root, prev_path]
if pending_node is None:
self.notify_filelist_changed(*args)
else:
pending_node.queue_move_notification(args)
else:
if pending_node is not None:
logging.debug(
"Parent node is processing, suppressing 'create from move' "
f"notification: {file_path}"
)
pending_node.reset_event("create_node", INOTIFY_BUNDLE_TIME)
if root == "gcodes":
self.parse_gcode_metadata(file_path)
return
self.sync_lock.add_pending_path("create_file", file_path)
if root == "gcodes":
if self.need_create_notify(file_path):
coro = self._finish_gcode_create_from_move(file_path)
self.queue_gcode_notification(coro)
else:
self.notify_filelist_changed("create_file", root, file_path)
elif evt.mask & iFlags.MODIFY:
self.sync_lock.add_pending_path("modify_file", file_path)
node.schedule_file_event(evt.name, "modify_file")
elif evt.mask & iFlags.CLOSE_WRITE:
logging.debug(f"Inotify writable file closed: {file_path}")
# Only process files that have been created or modified
node.complete_file_write(evt.name)
async def _finish_gcode_move(
self,
root: str,
prev_root: str,
file_path: str,
prev_path: str,
pending_node: Optional[InotifyNode],
move_result: Union[bool, Awaitable]
) -> None:
if not isinstance(move_result, bool):
await move_result
elif not move_result:
# Unable to move, metadata needs parsing
mevt = self.parse_gcode_metadata(file_path)
await mevt.wait()
args = ["move_file", root, file_path, prev_root, prev_path]
if pending_node is None:
self.notify_filelist_changed(*args)
else:
pending_node.queue_move_notification(args)
async def _finish_gcode_create_from_move(self, file_path: str) -> None:
mevt = self.parse_gcode_metadata(file_path)
await mevt.wait()
self.notify_filelist_changed("create_file", "gcodes", file_path)
self.clear_processing_file(file_path)
def need_create_notify(self, file_path: str) -> bool:
# We don't want to emit duplicate notifications, which may occur
# during metadata processing if the file needs to undergo object
# processing.
ext = os.path.splitext(file_path)[1].lower()
if ext == ".ufp":
# Queue the ufp file for parsing and return False, we do not
# want to notify the ufp since it will be removed.
self.parse_gcode_metadata(file_path)
return False
elif ext not in VALID_GCODE_EXTS:
return True
rel_path = self.file_manager.get_relative_path("gcodes", file_path)
if (
self.gcode_metadata.is_file_processing(rel_path) and
rel_path in self.processing_gcode_files
):
logging.debug(
f"Inotify file create event received for file '{rel_path}' during "
f"metadata processing. Suppressing notification."
)
return False
self.processing_gcode_files.add(rel_path)
return True
def clear_processing_file(self, file_path: str) -> None:
rel_path = self.file_manager.get_relative_path("gcodes", file_path)
self.processing_gcode_files.discard(rel_path)
def queue_gcode_notification(self, coro: Coroutine) -> None:
self.pending_coroutines.append(coro)
if self._gc_notify_task is None:
self._gc_notify_task = self.event_loop.create_task(
self._process_gcode_notifications()
)
async def _process_gcode_notifications(self) -> None:
while self.pending_coroutines:
coro = self.pending_coroutines.pop(0)
await coro
self._gc_notify_task = None
def notify_filelist_changed(self,
action: str,
root: str,
full_path: str,
source_root: Optional[str] = None,
source_path: Optional[str] = None
) -> None:
rel_path = self.file_manager.get_relative_path(root, full_path)
sync_fut = self.sync_lock.check_in_request(action, full_path)
file_info: Dict[str, Any] = {'size': 0, 'modified': 0, "permissions": ""}
if os.path.exists(full_path):
try:
file_info = self.file_manager.get_path_info(full_path, root)
except Exception:
logging.debug(
f"Invalid Filelist Notification Request, root: {root}, "
f"path: {full_path} - Failed to get path info")
return
elif action not in ["delete_file", "delete_dir"]:
logging.debug(
f"Invalid Filelist Notification Request, root: {root}, "
f"path: {full_path} - Action {action} received for file "
"that does not exit"
)
return
file_info['path'] = rel_path
file_info['root'] = root
result = {'action': action, 'item': file_info}
if source_path is not None and source_root is not None:
src_rel_path = self.file_manager.get_relative_path(
source_root, source_path)
result['source_item'] = {'path': src_rel_path, 'root': source_root}
key = f"{action}-{root}-{rel_path}"
self.event_loop.create_task(
self._finish_notify(result, sync_fut, key)
)
async def _finish_notify(
self,
result: Dict[str, Any],
sync_fut: Optional[asyncio.Future],
notify_key: str
) -> None:
if sync_fut is not None:
logging.debug(f"Syncing notification: {notify_key}")
await sync_fut
self.file_manager.cancel_notification(notify_key)
await asyncio.sleep(.005)
self.server.send_event("file_manager:filelist_changed", result)
def close(self) -> None:
while self.pending_coroutines:
coro = self.pending_coroutines.pop(0)
coro.close()
if self._gc_notify_task is not None:
self._gc_notify_task.cancel()
self.event_loop.remove_reader(self.inotify.fileno())
for watch in self.watched_nodes.keys():
try:
self.inotify.rm_watch(watch)
except Exception:
pass
METADATA_NAMESPACE = "gcode_metadata"
METADATA_VERSION = 3
class MetadataStorage:
def __init__(self,
config: ConfigHelper,
db: DBComp
) -> None:
self.server = config.get_server()
self.enable_object_proc = config.getboolean(
'enable_object_processing', False)
self.gc_path = ""
db.register_local_namespace(METADATA_NAMESPACE)
self.mddb = db.wrap_namespace(
METADATA_NAMESPACE, parse_keys=False)
version = db.get_item(
"moonraker", "file_manager.metadata_version", 0).result()
if version != METADATA_VERSION:
# Clear existing metadata when version is bumped
self.mddb.clear()
db.insert_item(
"moonraker", "file_manager.metadata_version",
METADATA_VERSION)
# Keep a local cache of the metadata. This allows for synchronous
# queries. Metadata is generally under 1KiB per entry, so even at
# 1000 gcode files we are using < 1MiB of additional memory.
# That said, in the future all components that access metadata should
# be refactored to do so asynchronously.
self.metadata: Dict[str, Any] = self.mddb.as_dict()
self.pending_requests: Dict[
str, Tuple[Dict[str, Any], asyncio.Event]] = {}
self.busy: bool = False
def prune_storage(self) -> None:
# Check for removed gcode files while moonraker was shutdown
if self.gc_path:
del_keys: List[str] = []
for fname in list(self.metadata.keys()):
fpath = os.path.join(self.gc_path, fname)
if not os.path.isfile(fpath):
del self.metadata[fname]
del_keys.append(fname)
elif "thumbnails" in self.metadata[fname]:
# Check for any stale data entries and remove them
need_sync = False
for thumb in self.metadata[fname]['thumbnails']:
if 'data' in thumb:
del thumb['data']
need_sync = True
if need_sync:
self.mddb[fname] = self.metadata[fname]
# Delete any removed keys from the database
if del_keys:
ret = self.mddb.delete_batch(del_keys).result()
self._remove_thumbs(ret)
pruned = '\n'.join(ret.keys())
if pruned:
logging.info(f"Pruned metadata for the following:\n"
f"{pruned}")
def update_gcode_path(self, path: str) -> None:
if path == self.gc_path:
return
if self.gc_path:
self.metadata.clear()
self.mddb.clear()
self.gc_path = path
def get(self,
key: str,
default: Optional[_T] = None
) -> Union[_T, Dict[str, Any]]:
return deepcopy(self.metadata.get(key, default))
def insert(self, key: str, value: Dict[str, Any]) -> None:
val = deepcopy(value)
self.metadata[key] = val
self.mddb[key] = val
def is_processing(self) -> bool:
return len(self.pending_requests) > 0
def is_file_processing(self, fname: str) -> bool:
return fname in self.pending_requests
def _has_valid_data(self,
fname: str,
path_info: Dict[str, Any]
) -> bool:
if path_info.get('ufp_path', None) is not None:
# UFP files always need processing
return False
mdata: Dict[str, Any]
mdata = self.metadata.get(fname, {'size': "", 'modified': 0})
for field in ['size', 'modified']:
if mdata[field] != path_info.get(field, None):
return False
return True
def remove_directory_metadata(self, dir_name: str) -> Optional[Awaitable]:
if dir_name[-1] != "/":
dir_name += "/"
del_items: Dict[str, Any] = {}
for fname in list(self.metadata.keys()):
if fname.startswith(dir_name):
md = self.metadata.pop(fname, None)
if md:
del_items[fname] = md
if del_items:
# Remove items from persistent storage
self.mddb.delete_batch(list(del_items.keys()))
eventloop = self.server.get_event_loop()
# Remove thumbs in a nother thread
return eventloop.run_in_thread(self._remove_thumbs, del_items)
return None
def remove_file_metadata(self, fname: str) -> Optional[Awaitable]:
md: Optional[Dict[str, Any]] = self.metadata.pop(fname, None)
if md is None:
return None
self.mddb.pop(fname, None)
eventloop = self.server.get_event_loop()
return eventloop.run_in_thread(self._remove_thumbs, {fname: md})
def _remove_thumbs(self, records: Dict[str, Dict[str, Any]]) -> None:
for fname, metadata in records.items():
# Delete associated thumbnails
fdir = os.path.dirname(os.path.join(self.gc_path, fname))
if "thumbnails" in metadata:
thumb: Dict[str, Any]
for thumb in metadata["thumbnails"]:
path: Optional[str] = thumb.get("relative_path", None)
if path is None:
continue
thumb_path = os.path.join(fdir, path)
if not os.path.isfile(thumb_path):
continue
try:
os.remove(thumb_path)
except Exception:
logging.debug(f"Error removing thumb at {thumb_path}")
def move_directory_metadata(self, prev_dir: str, new_dir: str) -> None:
if prev_dir[-1] != "/":
prev_dir += "/"
moved: List[Tuple[str, str, Dict[str, Any]]] = []
for prev_fname in list(self.metadata.keys()):
if prev_fname.startswith(prev_dir):
new_fname = os.path.join(new_dir, prev_fname[len(prev_dir):])
md: Optional[Dict[str, Any]]
md = self.metadata.pop(prev_fname, None)
if md is None:
continue
self.metadata[new_fname] = md
moved.append((prev_fname, new_fname, md))
if moved:
source = [m[0] for m in moved]
dest = [m[1] for m in moved]
self.mddb.move_batch(source, dest)
# It shouldn't be necessary to move the thumbnails
# as they would be moved with the parent directory
def move_file_metadata(
self, prev_fname: str, new_fname: str
) -> Union[bool, Awaitable]:
metadata: Optional[Dict[str, Any]]
metadata = self.metadata.pop(prev_fname, None)
if metadata is None:
# If this move overwrites an existing file it is necessary
# to rescan which requires that we remove any existing
# metadata.
if self.metadata.pop(new_fname, None) is not None:
self.mddb.pop(new_fname, None)
return False
self.metadata[new_fname] = metadata
self.mddb.move_batch([prev_fname], [new_fname])
return self._move_thumbnails([(prev_fname, new_fname, metadata)])
async def _move_thumbnails(
self, records: List[Tuple[str, str, Dict[str, Any]]]
) -> None:
eventloop = self.server.get_event_loop()
for (prev_fname, new_fname, metadata) in records:
prev_dir = os.path.dirname(os.path.join(self.gc_path, prev_fname))
new_dir = os.path.dirname(os.path.join(self.gc_path, new_fname))
if "thumbnails" in metadata:
thumb: Dict[str, Any]
for thumb in metadata["thumbnails"]:
path: Optional[str] = thumb.get("relative_path", None)
if path is None:
continue
thumb_path = os.path.join(prev_dir, path)
if not os.path.isfile(thumb_path):
continue
new_path = os.path.join(new_dir, path)
new_parent = os.path.dirname(new_path)
try:
if not os.path.exists(new_parent):
os.mkdir(new_parent)
# Wait for inotify to register the node before the move
await asyncio.sleep(.2)
await eventloop.run_in_thread(
shutil.move, thumb_path, new_path
)
except asyncio.CancelledError:
raise
except Exception:
logging.exception(
f"Error moving thumb from {thumb_path} to {new_path}"
)
def parse_metadata(self,
fname: str,
path_info: Dict[str, Any]
) -> asyncio.Event:
if fname in self.pending_requests:
return self.pending_requests[fname][1]
mevt = asyncio.Event()
ext = os.path.splitext(fname)[1]
if (
ext not in VALID_GCODE_EXTS or
self._has_valid_data(fname, path_info)
):
# request already pending or not necessary
mevt.set()
return mevt
self.pending_requests[fname] = (path_info, mevt)
if self.busy:
return mevt
self.busy = True
event_loop = self.server.get_event_loop()
event_loop.register_callback(self._process_metadata_update)
return mevt
async def _process_metadata_update(self) -> None:
while self.pending_requests:
fname, (path_info, mevt) = \
list(self.pending_requests.items())[0]
if self._has_valid_data(fname, path_info):
mevt.set()
continue
ufp_path: Optional[str] = path_info.get('ufp_path', None)
retries = 3
while retries:
try:
await self._run_extract_metadata(fname, ufp_path)
except Exception:
logging.exception("Error running extract_metadata.py")
retries -= 1
else:
break
else:
if ufp_path is None:
self.metadata[fname] = {
'size': path_info.get('size', 0),
'modified': path_info.get('modified', 0),
'print_start_time': None,
'job_id': None
}
self.mddb[fname] = self.metadata[fname]
logging.info(
f"Unable to extract medatadata from file: {fname}")
self.pending_requests.pop(fname, None)
mevt.set()
self.busy = False
async def _run_extract_metadata(self,
filename: str,
ufp_path: Optional[str]
) -> None:
# Escape single quotes in the file name so that it may be
# properly loaded
filename = filename.replace("\"", "\\\"")
cmd = " ".join([sys.executable, METADATA_SCRIPT, "-p",
self.gc_path, "-f", f"\"{filename}\""])
timeout = 10.
if ufp_path is not None and os.path.isfile(ufp_path):
timeout = 300.
ufp_path.replace("\"", "\\\"")
cmd += f" -u \"{ufp_path}\""
if self.enable_object_proc:
timeout = 300.
cmd += " --check-objects"
shell_cmd: SCMDComp = self.server.lookup_component('shell_command')
scmd = shell_cmd.build_shell_command(cmd, log_stderr=True)
result = await scmd.run_with_response(timeout=timeout)
try:
decoded_resp: Dict[str, Any] = json.loads(result.strip())
except Exception:
logging.debug(f"Invalid metadata response:\n{result}")
raise
path: str = decoded_resp['file']
metadata: Dict[str, Any] = decoded_resp['metadata']
if not metadata:
# This indicates an error, do not add metadata for this
raise self.server.error("Unable to extract metadata")
metadata.update({'print_start_time': None, 'job_id': None})
self.metadata[path] = metadata
self.mddb[path] = metadata
def load_component(config: ConfigHelper) -> FileManager:
return FileManager(config)