FishServer/fish_api/app/settings.py

from __future__ import annotations

from functools import lru_cache
from pathlib import Path
from typing import Optional

from pydantic import Field, field_validator, model_validator
from pydantic_settings import BaseSettings, SettingsConfigDict


def fish_repo_root() -> Path:
    # fish_api/app/settings.py -> parent[2] = repo root (contains FishMeasure/, fish_api/)
    return Path(__file__).resolve().parents[2]


def _default_stream_tmp() -> Path:
    return fish_repo_root() / "fish_api" / ".data" / "ingest"


def _default_media_root() -> Path:
    return fish_repo_root() / "fish_api" / ".data" / "media"


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        extra="ignore",
    )

    public_base_url: str = "http://127.0.0.1:8000"

    ingest_api_key: str = ""

    stream_tmp_dir: Path = Field(default_factory=_default_stream_tmp)
    media_root: Path = Field(default_factory=_default_media_root)

    fish_measure_root: Path = fish_repo_root() / "FishMeasure"
    fish_action_root: Path = fish_repo_root() / "FishAction"

    measure_output_root: Path = fish_repo_root() / "FishMeasure" / "output_weight_estimator"

    python_fish_measure: str = ""
    python_fish_action: str = ""

    yolo_model: Optional[str] = None
    weight_checkpoint: Optional[str] = None

    sam_device: str = "cuda"
    predict_conf: float = 0.5
    predict_imgsz: int = 640
    predict_max_frames: int = 0
    predict_frame_stride: int = 1

    action_checkpoint: Optional[str] = None
    action_clips_per_video: int = 8
    action_batch_size: int = 4
    action_num_workers: int = 2

    #: 非空时由 fish_api 在后台持续扫描该目录中的新 MP4 并跑 FishAction（与 ingest 共用 app_state）
    action_watch_dir: Optional[Path] = None
    action_watch_poll_interval: float = Field(default=2.0, ge=0.1)
    action_watch_stable_polls: int = Field(default=3, ge=1)
    action_watch_recursive: bool = False
    #: 默认：<action_watch_dir>/.fishaction_watch_processed.json
    action_watch_state_file: Optional[Path] = None
    action_watch_use_state_file: bool = True

    #: 非空时后台持续扫描该目录中的新 .svo2 并跑 FishMeasure（与 ingest 共用 app_state）
    measure_watch_dir: Optional[Path] = None
    measure_watch_poll_interval: float = Field(default=2.0, ge=0.1)
    measure_watch_stable_polls: int = Field(default=3, ge=1)
    measure_watch_recursive: bool = False
    measure_watch_state_file: Optional[Path] = None
    measure_watch_use_state_file: bool = True

    default_fish_species: str = "大黄鱼"

    @field_validator(
        "action_watch_dir",
        "action_watch_state_file",
        "measure_watch_dir",
        "measure_watch_state_file",
        mode="before",
    )
    @classmethod
    def _empty_str_path_none(cls, v: object) -> object:
        if v is None:
            return None
        if isinstance(v, str) and not v.strip():
            return None
        return v

    @model_validator(mode="after")
    def _default_paths(self) -> "Settings":
        if not self.yolo_model:
            object.__setattr__(
                self,
                "yolo_model",
                str(
                    self.fish_measure_root
                    / "runs/train/fish_detection_20251127_104658/weights/best.pt"
                ),
            )
        if not self.weight_checkpoint:
            object.__setattr__(
                self,
                "weight_checkpoint",
                str(
                    self.fish_measure_root
                    / "weight_estimator/runs/dgcnn_20260312_171043/best.pt"
                ),
            )
        if not self.action_checkpoint:
            object.__setattr__(
                self,
                "action_checkpoint",
                str(self.fish_action_root / "checkpoints/ptv_x3d_m/checkpoint_best.pt"),
            )
        return self


@lru_cache
def get_settings() -> Settings:
    return Settings()