nndeploy.server.server

Classes

NnDeployServer(args, job_mp_queue, ...)
SPAStaticFiles(*[, directory, packages, ...])

class nndeploy.server.server.SPAStaticFiles(*, directory: Optional[Union[str, PathLike[str]]] = None, packages: list[str | tuple[str, str]] | None = None, html: bool = False, check_dir: bool = True, follow_symlink: bool = False)

基类：StaticFiles

async get_response(path, scope)

Returns an HTTP response, given the incoming path, method and request headers.

__init__(*, directory: Optional[Union[str, PathLike[str]]] = None, packages: list[str | tuple[str, str]] | None = None, html: bool = False, check_dir: bool = True, follow_symlink: bool = False) → None

async check_config() → None

Perform a one-off configuration check that StaticFiles is actually pointed at a directory, so that we can raise loud errors rather than just returning 404 responses.

file_response(full_path: Union[str, PathLike[str]], stat_result: stat_result, scope: MutableMapping[str, Any], status_code: int = 200) → Response

get_directories(directory: Optional[Union[str, PathLike[str]]] = None, packages: list[str | tuple[str, str]] | None = None) → list[Union[str, os.PathLike[str]]]

Given directory and packages arguments, return a list of all the directories that should be used for serving static files from.

get_path(scope: MutableMapping[str, Any]) → str

Given the ASGI scope, return the path string to serve up, with OS specific path separators, and any ‘..’, ‘.’ components removed.

is_not_modified(response_headers: Headers, request_headers: Headers) → bool

Given the request and response headers, return True if an HTTP “Not Modified” response could be returned instead.

lookup_path(path: str) → tuple[str, os.stat_result | None]

class nndeploy.server.server.NnDeployServer(args, job_mp_queue, plugin_update_q, cancel_event_queue)

基类：object

instance: NnDeployServer = None

__init__(args, job_mp_queue, plugin_update_q, cancel_event_queue)

read_json_recursive(root: Path) → list[dict]

notify_download_done(task_id: str, success: bool, result: dict | None, error: str | None)

notify_task_progress(task_id: str, status_dict: dict)

notify_task_done(task_id: str, status: ExecutionStatus, results: Dict, time_profile_map: Dict)

notify_system_event(event: str, data: dict | None = None) → None

notify system event to all connected websocket clients

notify_download_progress(task_id: str, ev: dict)

Broadcast a single download progress event to all sockets bound to task_id.

Expected ev example (from DownloadProgressHandler): {

“phase”: “progress” | “done” | “error”, “filename”: “detect/yolo11s.sim.onnx”, “downloaded”: 1048576, “total”: 37958400, “percent”: 2.76, “elapsed”: 1.23, “logger”: “model-download” | “modelscope”, … # (optional) task_id, bps, eta, etc.

}