739 lines
20 KiB
Python
739 lines
20 KiB
Python
# backend\ai_server.py
|
|
|
|
import json
|
|
import os
|
|
import secrets
|
|
from pathlib import Path
|
|
from typing import List, Optional
|
|
|
|
from fastapi import Depends, FastAPI, HTTPException, Request, status
|
|
from pydantic import BaseModel
|
|
from ultralytics import YOLO
|
|
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent
|
|
KEYPOINT_NAMES = [
|
|
"nose",
|
|
"left_eye", "right_eye",
|
|
"left_ear", "right_ear",
|
|
"left_shoulder", "right_shoulder",
|
|
"left_elbow", "right_elbow",
|
|
"left_wrist", "right_wrist",
|
|
"left_hip", "right_hip",
|
|
"left_knee", "right_knee",
|
|
"left_ankle", "right_ankle",
|
|
]
|
|
|
|
NO_SEX_POSITION_LABEL = "keine"
|
|
NO_SEX_POSITION_ALIASES = {
|
|
"",
|
|
NO_SEX_POSITION_LABEL,
|
|
}
|
|
|
|
|
|
def normalize_sex_position_label(value) -> str:
|
|
clean = str(value or "").strip().lower()
|
|
if clean in NO_SEX_POSITION_ALIASES:
|
|
return NO_SEX_POSITION_LABEL
|
|
return clean
|
|
|
|
|
|
def is_no_sex_position_label(value) -> bool:
|
|
return normalize_sex_position_label(value) == NO_SEX_POSITION_LABEL
|
|
|
|
|
|
def normalize_sex_position_labels(values) -> set[str]:
|
|
labels = set()
|
|
has_no_position = False
|
|
|
|
for value in values or []:
|
|
clean = normalize_sex_position_label(value)
|
|
if is_no_sex_position_label(clean):
|
|
has_no_position = True
|
|
continue
|
|
if clean:
|
|
labels.add(clean)
|
|
|
|
if has_no_position:
|
|
labels.add(NO_SEX_POSITION_LABEL)
|
|
|
|
return labels
|
|
|
|
|
|
def existing_file(path: Path) -> Optional[Path]:
|
|
try:
|
|
if path.exists() and path.is_file() and path.stat().st_size > 0:
|
|
return path
|
|
except OSError:
|
|
pass
|
|
return None
|
|
|
|
|
|
def resolve_training_root() -> Path:
|
|
env_root = os.environ.get("TRAINING_ROOT", "").strip()
|
|
if env_root:
|
|
root = Path(env_root).expanduser().resolve()
|
|
root.mkdir(parents=True, exist_ok=True)
|
|
return root
|
|
|
|
candidates = [
|
|
# Wenn ai_server.py aus backend/ läuft:
|
|
BASE_DIR / "generated" / "training",
|
|
|
|
# Wenn ai_server.py aus backend/ml/ laufen würde:
|
|
BASE_DIR.parent / "generated" / "training",
|
|
|
|
# Wenn ai_server.py embedded aus Temp läuft, aber backendRoot als cwd gesetzt wurde:
|
|
Path.cwd() / "generated" / "training",
|
|
|
|
# Wenn Working Directory Projektroot ist:
|
|
Path.cwd() / "backend" / "generated" / "training",
|
|
]
|
|
|
|
for root in candidates:
|
|
if (
|
|
existing_file(root / "detection_labels.json")
|
|
or existing_file(root / "detector" / "model" / "best.pt")
|
|
):
|
|
root.mkdir(parents=True, exist_ok=True)
|
|
return root.resolve()
|
|
|
|
# Fallback: Server soll trotzdem starten.
|
|
root = (Path.cwd() / "generated" / "training").resolve()
|
|
root.mkdir(parents=True, exist_ok=True)
|
|
return root
|
|
|
|
|
|
TRAINING_ROOT = resolve_training_root()
|
|
DEFAULT_MODEL_PATH = TRAINING_ROOT / "detector" / "model" / "best.pt"
|
|
DEFAULT_POSE_MODEL_PATH = TRAINING_ROOT / "pose" / "model" / "best.pt"
|
|
|
|
|
|
def resolve_detection_labels_path() -> Path:
|
|
env_path = os.environ.get("DETECTION_LABELS_PATH", "").strip()
|
|
if env_path:
|
|
p = Path(env_path).expanduser().resolve()
|
|
if existing_file(p):
|
|
return p
|
|
raise RuntimeError(f"DETECTION_LABELS_PATH not found: {p}")
|
|
|
|
p = TRAINING_ROOT / "detection_labels.json"
|
|
if existing_file(p):
|
|
return p.resolve()
|
|
|
|
raise RuntimeError(f"detection_labels.json not found: {p}")
|
|
|
|
|
|
def resolve_model_path() -> str:
|
|
env_path = os.environ.get("YOLO_MODEL", "").strip()
|
|
if env_path:
|
|
p = Path(env_path).expanduser().resolve()
|
|
if existing_file(p):
|
|
return str(p)
|
|
raise RuntimeError(f"YOLO_MODEL not found: {p}")
|
|
|
|
if existing_file(DEFAULT_MODEL_PATH):
|
|
return str(DEFAULT_MODEL_PATH)
|
|
|
|
raise RuntimeError(f"YOLO model not found: {DEFAULT_MODEL_PATH}")
|
|
|
|
|
|
def resolve_pose_model_path() -> Optional[Path]:
|
|
env_path = os.environ.get("YOLO_POSE_MODEL", "").strip()
|
|
if env_path:
|
|
p = Path(env_path).expanduser().resolve()
|
|
if existing_file(p):
|
|
return p
|
|
raise RuntimeError(f"YOLO_POSE_MODEL not found: {p}")
|
|
|
|
if existing_file(DEFAULT_POSE_MODEL_PATH):
|
|
return DEFAULT_POSE_MODEL_PATH.resolve()
|
|
|
|
return None
|
|
|
|
|
|
# Server darf auch ohne Labels/Model starten.
|
|
DETECTION_LABELS_PATH: Optional[Path] = None
|
|
|
|
LABEL_GROUPS = {
|
|
"people": set(),
|
|
"sexPositions": {NO_SEX_POSITION_LABEL},
|
|
"bodyParts": set(),
|
|
"objects": set(),
|
|
"clothing": set(),
|
|
}
|
|
|
|
BODY_LABELS = LABEL_GROUPS["bodyParts"]
|
|
OBJECT_LABELS = LABEL_GROUPS["objects"]
|
|
CLOTHING_LABELS = LABEL_GROUPS["clothing"]
|
|
POSITION_LABELS = set()
|
|
|
|
PERSON_LABELS = {
|
|
"person_male",
|
|
"person_female",
|
|
}
|
|
|
|
_MODEL_PATH = ""
|
|
_MODEL_ERROR = ""
|
|
_POSE_MODEL_PATH = ""
|
|
_POSE_MODEL_ERROR = ""
|
|
_LABEL_ERROR = ""
|
|
|
|
_DEVICE = os.environ.get("YOLO_DEVICE", "")
|
|
_CONF = float(os.environ.get("YOLO_CONF", "0.25"))
|
|
_POSE_CONF = float(os.environ.get("YOLO_POSE_CONF", "0.30"))
|
|
_BATCH = int(os.environ.get("YOLO_BATCH", "16"))
|
|
_IMGSZ = int(os.environ.get("YOLO_IMGSZ", "640"))
|
|
_HALF = os.environ.get("YOLO_HALF", "0").lower() in {"1", "true", "yes", "on"}
|
|
|
|
model = None
|
|
pose_model = None
|
|
|
|
app = FastAPI()
|
|
|
|
AI_SERVER_TOKEN = os.environ.get("AI_SERVER_TOKEN", "").strip()
|
|
AI_SERVER_AUTH_REQUIRED = os.environ.get("AI_SERVER_AUTH_REQUIRED", "1").strip().lower() not in {
|
|
"0",
|
|
"false",
|
|
"no",
|
|
"off",
|
|
}
|
|
|
|
|
|
def require_ai_server_auth(request: Request) -> None:
|
|
if not AI_SERVER_AUTH_REQUIRED:
|
|
return
|
|
|
|
if not AI_SERVER_TOKEN:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="AI server auth token not configured",
|
|
)
|
|
|
|
auth = request.headers.get("authorization", "").strip()
|
|
header_token = request.headers.get("x-ai-server-token", "").strip()
|
|
|
|
provided = ""
|
|
|
|
if auth.lower().startswith("bearer "):
|
|
provided = auth[7:].strip()
|
|
elif header_token:
|
|
provided = header_token
|
|
|
|
if not provided or not secrets.compare_digest(provided, AI_SERVER_TOKEN):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="unauthorized",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
|
|
class PredictBatchRequest(BaseModel):
|
|
paths: List[str]
|
|
detectorOnly: bool = False
|
|
imageSize: int = 640
|
|
model: Optional[str] = None
|
|
|
|
|
|
def empty_prediction(source: str = "model_missing") -> dict:
|
|
return {
|
|
"modelAvailable": False,
|
|
"source": source,
|
|
"sexPosition": NO_SEX_POSITION_LABEL,
|
|
"sexPositionScore": 0.0,
|
|
"peoplePresent": [],
|
|
"bodyPartsPresent": [],
|
|
"objectsPresent": [],
|
|
"clothingPresent": [],
|
|
"boxes": [],
|
|
"persons": [],
|
|
}
|
|
|
|
|
|
def load_label_groups_safe() -> None:
|
|
global DETECTION_LABELS_PATH
|
|
global LABEL_GROUPS
|
|
global BODY_LABELS
|
|
global OBJECT_LABELS
|
|
global CLOTHING_LABELS
|
|
global POSITION_LABELS
|
|
global PERSON_LABELS
|
|
global _LABEL_ERROR
|
|
|
|
try:
|
|
path = resolve_detection_labels_path()
|
|
DETECTION_LABELS_PATH = path
|
|
|
|
with path.open("r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
|
|
LABEL_GROUPS = {
|
|
"people": set(
|
|
str(x).strip().lower()
|
|
for x in data.get("people", [])
|
|
if str(x).strip()
|
|
),
|
|
"sexPositions": normalize_sex_position_labels(data.get("sexPositions", [])),
|
|
"bodyParts": set(
|
|
str(x).strip().lower()
|
|
for x in data.get("bodyParts", [])
|
|
if str(x).strip()
|
|
),
|
|
"objects": set(
|
|
str(x).strip().lower()
|
|
for x in data.get("objects", [])
|
|
if str(x).strip()
|
|
),
|
|
"clothing": set(
|
|
str(x).strip().lower()
|
|
for x in data.get("clothing", [])
|
|
if str(x).strip()
|
|
),
|
|
}
|
|
|
|
if not LABEL_GROUPS["sexPositions"]:
|
|
LABEL_GROUPS["sexPositions"] = {NO_SEX_POSITION_LABEL}
|
|
|
|
_LABEL_ERROR = ""
|
|
|
|
except Exception as exc:
|
|
DETECTION_LABELS_PATH = None
|
|
_LABEL_ERROR = str(exc)
|
|
|
|
LABEL_GROUPS = {
|
|
"people": set(),
|
|
"sexPositions": {NO_SEX_POSITION_LABEL},
|
|
"bodyParts": set(),
|
|
"objects": set(),
|
|
"clothing": set(),
|
|
}
|
|
|
|
BODY_LABELS = LABEL_GROUPS["bodyParts"]
|
|
OBJECT_LABELS = LABEL_GROUPS["objects"]
|
|
CLOTHING_LABELS = LABEL_GROUPS["clothing"]
|
|
|
|
POSITION_LABELS = {
|
|
label for label in LABEL_GROUPS["sexPositions"]
|
|
if label and not is_no_sex_position_label(label)
|
|
}
|
|
|
|
PERSON_LABELS = {
|
|
label for label in LABEL_GROUPS["people"]
|
|
if label
|
|
}
|
|
|
|
|
|
def get_model():
|
|
global model
|
|
global _MODEL_PATH
|
|
global _MODEL_ERROR
|
|
|
|
if model is not None:
|
|
return model
|
|
|
|
try:
|
|
path = resolve_model_path()
|
|
loaded = YOLO(path)
|
|
|
|
model = loaded
|
|
_MODEL_PATH = path
|
|
_MODEL_ERROR = ""
|
|
|
|
# Labels erst laden, wenn Inference wirklich gebraucht wird.
|
|
load_label_groups_safe()
|
|
|
|
return model
|
|
|
|
except Exception as exc:
|
|
model = None
|
|
_MODEL_PATH = ""
|
|
_MODEL_ERROR = str(exc)
|
|
return None
|
|
|
|
|
|
def get_pose_model():
|
|
global pose_model
|
|
global _POSE_MODEL_PATH
|
|
global _POSE_MODEL_ERROR
|
|
|
|
if pose_model is not None:
|
|
return pose_model
|
|
|
|
try:
|
|
path = resolve_pose_model_path()
|
|
if path is None:
|
|
pose_model = None
|
|
_POSE_MODEL_PATH = ""
|
|
_POSE_MODEL_ERROR = ""
|
|
return None
|
|
|
|
loaded = YOLO(str(path))
|
|
|
|
pose_model = loaded
|
|
_POSE_MODEL_PATH = str(path)
|
|
_POSE_MODEL_ERROR = ""
|
|
|
|
return pose_model
|
|
|
|
except Exception as exc:
|
|
pose_model = None
|
|
_POSE_MODEL_PATH = ""
|
|
_POSE_MODEL_ERROR = str(exc)
|
|
return None
|
|
|
|
|
|
def scored(label: str, score: float) -> dict:
|
|
return {
|
|
"label": label,
|
|
"score": float(score),
|
|
}
|
|
|
|
|
|
def best_score(items: list[dict], label: str, score: float) -> None:
|
|
for item in items:
|
|
if item["label"] == label:
|
|
if score > item["score"]:
|
|
item["score"] = float(score)
|
|
return
|
|
|
|
items.append(scored(label, score))
|
|
|
|
|
|
def prediction_from_result(result) -> dict:
|
|
names = result.names or {}
|
|
|
|
boxes_out = []
|
|
people_present = []
|
|
body_parts = []
|
|
objects = []
|
|
clothing = []
|
|
|
|
if result.boxes is not None:
|
|
xywhn = result.boxes.xywhn.cpu().tolist()
|
|
cls_values = result.boxes.cls.cpu().tolist()
|
|
conf_values = result.boxes.conf.cpu().tolist()
|
|
|
|
for box_xywhn, cls_id, conf in zip(xywhn, cls_values, conf_values):
|
|
label = str(names.get(int(cls_id), int(cls_id))).strip().lower()
|
|
score = float(conf)
|
|
|
|
if not label:
|
|
continue
|
|
|
|
cx, cy, w, h = [float(v) for v in box_xywhn]
|
|
|
|
x = max(0.0, min(1.0, cx - w / 2.0))
|
|
y = max(0.0, min(1.0, cy - h / 2.0))
|
|
w = max(0.0, min(1.0 - x, w))
|
|
h = max(0.0, min(1.0 - y, h))
|
|
|
|
is_person = label in PERSON_LABELS
|
|
is_body = label in BODY_LABELS
|
|
is_object = label in OBJECT_LABELS
|
|
is_clothing = label in CLOTHING_LABELS
|
|
|
|
if not (is_person or is_body or is_object or is_clothing):
|
|
continue
|
|
|
|
boxes_out.append({
|
|
"label": label,
|
|
"score": score,
|
|
"x": x,
|
|
"y": y,
|
|
"w": w,
|
|
"h": h,
|
|
})
|
|
|
|
if is_person:
|
|
best_score(people_present, label, score)
|
|
|
|
if is_body:
|
|
best_score(body_parts, label, score)
|
|
|
|
if is_object:
|
|
best_score(objects, label, score)
|
|
|
|
if is_clothing:
|
|
best_score(clothing, label, score)
|
|
|
|
return {
|
|
"modelAvailable": True,
|
|
"source": f"yolo-server:{Path(_MODEL_PATH).name}",
|
|
"sexPosition": NO_SEX_POSITION_LABEL,
|
|
"sexPositionScore": 0.0,
|
|
"peoplePresent": people_present,
|
|
"bodyPartsPresent": body_parts,
|
|
"objectsPresent": objects,
|
|
"clothingPresent": clothing,
|
|
"boxes": boxes_out,
|
|
"persons": [],
|
|
}
|
|
|
|
|
|
def pose_persons_from_result(result) -> list[dict]:
|
|
names = result.names or {}
|
|
persons = []
|
|
|
|
if result.boxes is None:
|
|
return persons
|
|
|
|
kpts_xyn = None
|
|
kpts_conf = None
|
|
|
|
if result.keypoints is not None:
|
|
try:
|
|
kpts_xyn = result.keypoints.xyn.cpu().tolist()
|
|
except Exception:
|
|
kpts_xyn = None
|
|
try:
|
|
kpts_conf = result.keypoints.conf.cpu().tolist()
|
|
except Exception:
|
|
kpts_conf = None
|
|
|
|
xywhn_values = result.boxes.xywhn.cpu().tolist()
|
|
cls_values = result.boxes.cls.cpu().tolist()
|
|
conf_values = result.boxes.conf.cpu().tolist()
|
|
|
|
for i, (box_xywhn, cls_id, conf) in enumerate(
|
|
zip(xywhn_values, cls_values, conf_values)
|
|
):
|
|
label = str(names.get(int(cls_id), int(cls_id))).strip().lower()
|
|
score = float(conf)
|
|
|
|
cx, cy, w, h = [float(v) for v in box_xywhn]
|
|
x = max(0.0, min(1.0, cx - w / 2.0))
|
|
y = max(0.0, min(1.0, cy - h / 2.0))
|
|
w = max(0.0, min(1.0 - x, w))
|
|
h = max(0.0, min(1.0 - y, h))
|
|
|
|
if w <= 0 or h <= 0:
|
|
continue
|
|
|
|
keypoints = []
|
|
if kpts_xyn is not None and i < len(kpts_xyn):
|
|
for ki, point in enumerate(kpts_xyn[i]):
|
|
if len(point) < 2:
|
|
continue
|
|
|
|
kconf = 0.0
|
|
if (
|
|
kpts_conf is not None
|
|
and i < len(kpts_conf)
|
|
and ki < len(kpts_conf[i])
|
|
):
|
|
kconf = float(kpts_conf[i][ki])
|
|
|
|
keypoints.append({
|
|
"name": KEYPOINT_NAMES[ki] if ki < len(KEYPOINT_NAMES) else str(ki),
|
|
"x": float(point[0]),
|
|
"y": float(point[1]),
|
|
"conf": kconf,
|
|
})
|
|
|
|
persons.append({
|
|
"label": label,
|
|
"score": score,
|
|
"box": {
|
|
"x": x,
|
|
"y": y,
|
|
"w": w,
|
|
"h": h,
|
|
},
|
|
"keypoints": keypoints,
|
|
})
|
|
|
|
return persons
|
|
|
|
|
|
def apply_pose_result_to_prediction(prediction: dict, result) -> dict:
|
|
persons = pose_persons_from_result(result)
|
|
if not persons:
|
|
return prediction
|
|
|
|
prediction["persons"] = persons
|
|
|
|
best_position = ""
|
|
best_score_value = 0.0
|
|
|
|
for person in persons:
|
|
label = str(person.get("label") or "").strip().lower()
|
|
score = float(person.get("score") or 0.0)
|
|
|
|
if is_no_sex_position_label(label) or label not in POSITION_LABELS:
|
|
continue
|
|
|
|
if score > best_score_value:
|
|
best_position = label
|
|
best_score_value = score
|
|
|
|
if best_position:
|
|
prediction["sexPosition"] = best_position
|
|
prediction["sexPositionScore"] = best_score_value
|
|
|
|
source = str(prediction.get("source") or "").strip()
|
|
prediction["source"] = f"{source}+yolo_pose" if source else "yolo_pose"
|
|
|
|
return prediction
|
|
|
|
|
|
def apply_pose_batch_to_predictions(paths: list[str], predictions: list[dict], imgsz: int) -> None:
|
|
global _POSE_MODEL_ERROR
|
|
|
|
current_pose_model = get_pose_model()
|
|
if current_pose_model is None:
|
|
return
|
|
|
|
try:
|
|
pose_results = current_pose_model.predict(
|
|
source=paths,
|
|
imgsz=imgsz,
|
|
conf=_POSE_CONF,
|
|
batch=_BATCH,
|
|
device=_DEVICE or None,
|
|
half=_HALF,
|
|
verbose=False,
|
|
)
|
|
except Exception as exc:
|
|
_POSE_MODEL_ERROR = str(exc)
|
|
return
|
|
|
|
for prediction, pose_result in zip(predictions, pose_results):
|
|
apply_pose_result_to_prediction(prediction, pose_result)
|
|
|
|
|
|
def pose_model_status() -> dict:
|
|
try:
|
|
expected = resolve_pose_model_path()
|
|
expected_text = str(expected) if expected else str(DEFAULT_POSE_MODEL_PATH)
|
|
exists = expected is not None
|
|
error = _POSE_MODEL_ERROR
|
|
except Exception as exc:
|
|
expected_text = str(DEFAULT_POSE_MODEL_PATH)
|
|
exists = False
|
|
error = str(exc)
|
|
|
|
return {
|
|
"poseModelAvailable": exists,
|
|
"poseModelLoaded": pose_model is not None,
|
|
"poseModel": _POSE_MODEL_PATH or (expected_text if exists else ""),
|
|
"poseModelError": error,
|
|
"expectedPoseModel": expected_text,
|
|
}
|
|
|
|
|
|
@app.post("/predict-batch", dependencies=[Depends(require_ai_server_auth)])
|
|
def predict_batch(req: PredictBatchRequest):
|
|
paths = [str(path).strip() for path in req.paths if str(path).strip()]
|
|
if not paths:
|
|
return {
|
|
"ok": False,
|
|
"predictions": [],
|
|
"error": "no paths supplied",
|
|
}
|
|
|
|
current_model = get_model()
|
|
if current_model is None:
|
|
return {
|
|
"ok": True,
|
|
"predictions": [empty_prediction("model_missing") for _ in paths],
|
|
"error": _MODEL_ERROR or f"YOLO model not found: {DEFAULT_MODEL_PATH}",
|
|
}
|
|
|
|
if DETECTION_LABELS_PATH is None or _LABEL_ERROR:
|
|
return {
|
|
"ok": False,
|
|
"predictions": [],
|
|
"error": f"detection labels missing: {_LABEL_ERROR}",
|
|
}
|
|
|
|
imgsz = int(req.imageSize or _IMGSZ or 640)
|
|
|
|
try:
|
|
results = current_model.predict(
|
|
source=paths,
|
|
imgsz=imgsz,
|
|
conf=_CONF,
|
|
batch=_BATCH,
|
|
device=_DEVICE or None,
|
|
half=_HALF,
|
|
verbose=False,
|
|
)
|
|
|
|
predictions = [prediction_from_result(result) for result in results]
|
|
|
|
if not req.detectorOnly:
|
|
apply_pose_batch_to_predictions(paths, predictions, imgsz)
|
|
|
|
return {
|
|
"ok": True,
|
|
"predictions": predictions,
|
|
}
|
|
|
|
except Exception as exc:
|
|
return {
|
|
"ok": False,
|
|
"predictions": [],
|
|
"error": str(exc),
|
|
}
|
|
|
|
|
|
@app.get("/health", dependencies=[Depends(require_ai_server_auth)])
|
|
def health():
|
|
current_model = get_model()
|
|
names = getattr(current_model, "names", {}) or {} if current_model is not None else {}
|
|
|
|
status_payload = {
|
|
"ok": True,
|
|
"ready": current_model is not None,
|
|
"modelAvailable": current_model is not None,
|
|
"model": _MODEL_PATH,
|
|
"modelError": _MODEL_ERROR,
|
|
"expectedModel": str(DEFAULT_MODEL_PATH),
|
|
"trainingRoot": str(TRAINING_ROOT),
|
|
"classCount": len(names),
|
|
"classes": list(names.values())[:80] if isinstance(names, dict) else names,
|
|
"labelConfig": str(DETECTION_LABELS_PATH) if DETECTION_LABELS_PATH else "",
|
|
"labelError": _LABEL_ERROR,
|
|
}
|
|
|
|
status_payload.update(pose_model_status())
|
|
return status_payload
|
|
|
|
@app.post("/reload", dependencies=[Depends(require_ai_server_auth)])
|
|
def reload_model():
|
|
global model
|
|
global pose_model
|
|
global _MODEL_PATH
|
|
global _MODEL_ERROR
|
|
global _POSE_MODEL_PATH
|
|
global _POSE_MODEL_ERROR
|
|
global DETECTION_LABELS_PATH
|
|
|
|
model = None
|
|
pose_model = None
|
|
_MODEL_PATH = ""
|
|
_MODEL_ERROR = ""
|
|
_POSE_MODEL_PATH = ""
|
|
_POSE_MODEL_ERROR = ""
|
|
DETECTION_LABELS_PATH = None
|
|
|
|
current_model = get_model()
|
|
names = getattr(current_model, "names", {}) or {} if current_model is not None else {}
|
|
|
|
status_payload = {
|
|
"ok": current_model is not None,
|
|
"ready": current_model is not None,
|
|
"modelAvailable": current_model is not None,
|
|
"model": _MODEL_PATH,
|
|
"modelError": _MODEL_ERROR,
|
|
"expectedModel": str(DEFAULT_MODEL_PATH),
|
|
"trainingRoot": str(TRAINING_ROOT),
|
|
"classCount": len(names),
|
|
"classes": list(names.values())[:80] if isinstance(names, dict) else names,
|
|
"labelConfig": str(DETECTION_LABELS_PATH) if DETECTION_LABELS_PATH else "",
|
|
"labelError": _LABEL_ERROR,
|
|
}
|
|
|
|
status_payload.update(pose_model_status())
|
|
return status_payload
|