mirror of
https://github.com/immich-app/immich.git
synced 2026-06-04 05:45:24 -04:00
renovate[bot]
408e1180ca
* chore(deps): update machine-learning * fix typing * fix deprecation log * no control socket --------- Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com> Co-authored-by: mertalev <101130780+mertalev@users.noreply.github.com>
130 lines
2.5 KiB
Python
130 lines
2.5 KiB
Python
from enum import Enum
|
|
from typing import Any, Literal, Protocol, TypeGuard, TypeVar
|
|
|
|
import numpy as np
|
|
import numpy.typing as npt
|
|
import orjson
|
|
from fastapi.responses import JSONResponse
|
|
from typing_extensions import TypedDict
|
|
|
|
|
|
class ORJSONResponse(JSONResponse):
|
|
def render(self, content: Any) -> bytes:
|
|
return orjson.dumps(content, option=orjson.OPT_SERIALIZE_NUMPY)
|
|
|
|
|
|
class StrEnum(str, Enum):
|
|
value: str
|
|
|
|
def __str__(self) -> str:
|
|
return self.value
|
|
|
|
|
|
class BoundingBox(TypedDict):
|
|
x1: int
|
|
y1: int
|
|
x2: int
|
|
y2: int
|
|
|
|
|
|
class ModelTask(StrEnum):
|
|
FACIAL_RECOGNITION = "facial-recognition"
|
|
SEARCH = "clip"
|
|
OCR = "ocr"
|
|
|
|
|
|
class ModelType(StrEnum):
|
|
DETECTION = "detection"
|
|
RECOGNITION = "recognition"
|
|
TEXTUAL = "textual"
|
|
VISUAL = "visual"
|
|
|
|
|
|
class ModelFormat(StrEnum):
|
|
ARMNN = "armnn"
|
|
ONNX = "onnx"
|
|
RKNN = "rknn"
|
|
|
|
|
|
class ModelSource(StrEnum):
|
|
INSIGHTFACE = "insightface"
|
|
MCLIP = "mclip"
|
|
OPENCLIP = "openclip"
|
|
PADDLE = "paddle"
|
|
|
|
|
|
class ModelPrecision(StrEnum):
|
|
FP16 = "FP16"
|
|
FP32 = "FP32"
|
|
|
|
|
|
ModelIdentity = tuple[ModelType, ModelTask]
|
|
|
|
|
|
class SessionNode(Protocol):
|
|
@property
|
|
def name(self) -> str | None: ...
|
|
|
|
@property
|
|
def shape(self) -> tuple[int, ...]: ...
|
|
|
|
|
|
class ModelSession(Protocol):
|
|
def run(
|
|
self,
|
|
output_names: list[str] | None,
|
|
input_feed: dict[str, npt.NDArray[np.float32]] | dict[str, npt.NDArray[np.int32]],
|
|
run_options: Any = None,
|
|
) -> list[npt.NDArray[np.float32]]: ...
|
|
|
|
def get_inputs(self) -> list[SessionNode]: ...
|
|
|
|
def get_outputs(self) -> list[SessionNode]: ...
|
|
|
|
|
|
class HasProfiling(Protocol):
|
|
profiling: dict[str, float]
|
|
|
|
|
|
class FaceDetectionOutput(TypedDict):
|
|
boxes: npt.NDArray[np.float32]
|
|
scores: npt.NDArray[np.float32]
|
|
landmarks: npt.NDArray[np.float32]
|
|
|
|
|
|
class DetectedFace(TypedDict):
|
|
boundingBox: BoundingBox
|
|
embedding: str
|
|
score: float
|
|
|
|
|
|
FacialRecognitionOutput = list[DetectedFace]
|
|
|
|
|
|
class PipelineEntry(TypedDict):
|
|
modelName: str
|
|
options: dict[str, Any]
|
|
|
|
|
|
PipelineRequest = dict[ModelTask, dict[ModelType, PipelineEntry]]
|
|
|
|
|
|
class InferenceEntry(TypedDict):
|
|
name: str
|
|
task: ModelTask
|
|
type: ModelType
|
|
options: dict[str, Any]
|
|
|
|
|
|
InferenceEntries = tuple[list[InferenceEntry], list[InferenceEntry]]
|
|
|
|
|
|
InferenceResponse = dict[ModelTask | Literal["imageHeight"] | Literal["imageWidth"], Any]
|
|
|
|
|
|
def has_profiling(obj: Any) -> TypeGuard[HasProfiling]:
|
|
return hasattr(obj, "profiling") and isinstance(obj.profiling, dict)
|
|
|
|
|
|
T = TypeVar("T")
|