Files
2026-07-29 15:01:28 +07:00

131 lines
3.8 KiB
Python

"""
Per-unit configuration for Detection by Case.
Model and line settings must match each production site.
"""
import os
from dataclasses import dataclass
from typing import Optional
@dataclass(frozen=True)
class UnitConfig:
site_name: str
model_path: str
class_ayam: str = "ayam"
class_talenan: str = "talenan"
cross_direction: str = "rtl"
line_x: Optional[int] = None
line_x_frac: float = 0.5
imgsz: int = 416
half: bool = True
conf: float = 0.3
device: int = 0
tracker: str = "bytetrack.yaml"
camera_name: str = "CASE"
object_label: str = "ayam-potong"
min_object_per_batch: int = 60
min_duration_per_batch: int = 60
def _env(key: str, default: str) -> str:
return os.getenv(key, default)
def _env_int_optional(key: str) -> Optional[int]:
raw = os.getenv(key)
if raw is None:
return None
raw = raw.strip()
if not raw:
return None
try:
value = int(raw)
except ValueError:
return None
return value if value >= 0 else None
def _env_float_01(key: str, default: float) -> float:
raw = os.getenv(key)
if raw is None:
return default
raw = raw.strip()
if not raw:
return default
try:
value = float(raw)
except ValueError:
return default
if value > 1.0 and value <= 100.0:
value = value / 100.0
if not 0.0 <= value <= 1.0:
return default
return value
def _env_direction(key: str, default: str = "rtl") -> str:
raw = os.getenv(key, default).strip().lower()
return raw if raw in {"rtl", "ltr", "both"} else default
def _build_unit(
site_name: str,
model_env: str,
model_default: str,
camera_name: str,
line_x_frac_default: float = 0.5,
line_x_default: Optional[int] = None,
) -> UnitConfig:
site_key = site_name.upper()
return UnitConfig(
site_name=site_name,
model_path=_env(model_env, model_default),
camera_name=camera_name,
line_x_frac=_env_float_01(f"CASE_{site_key}_LINE_X_FRAC", line_x_frac_default),
line_x=_env_int_optional(f"CASE_{site_key}_LINE_X") or line_x_default,
cross_direction=_env_direction(f"CASE_{site_key}_CROSS_DIRECTION", "rtl"),
imgsz=int(_env("CASE_IMGSZ", "416")),
half=_env("CASE_HALF", "true").lower() == "true",
conf=float(_env("CASE_CONF", "0.3")),
device=int(_env("CASE_DEVICE", "0")),
tracker=_env("CASE_TRACKER", "bytetrack.yaml"),
min_object_per_batch=int(_env("CASE_MIN_OBJECT_PER_BATCH", "1")),
min_duration_per_batch=int(_env("CASE_MIN_DURATION_PER_BATCH", "1")),
)
UNITS: dict[str, UnitConfig] = {
"Salatiga": _build_unit(
"Salatiga",
"CASE_SALATIGA_MODEL",
"/media/jetson/DATA/yolo11n-salatiga.engine",
"SL1",
line_x_frac_default=0.5,
),
"Cicalengka": _build_unit(
"Cicalengka",
"CASE_CICALENGKA_MODEL",
"/media/jetson/DATA/yolo11n-cicalengka.engine",
"CC1",
line_x_frac_default=0.5,
),
}
ALLOWED_UNITS = tuple(UNITS.keys())
MAX_UPLOAD_BYTES = int(_env("CASE_MAX_UPLOAD_MB", "500")) * 1024 * 1024
ALLOWED_EXTENSIONS = {".mp4", ".avi", ".mov"}
TEMP_DIR = _env("CASE_TEMP_DIR", os.path.join(os.path.sep, "tmp", "nobar-ayam"))
APP_HOST = _env("CASE_HOST", "0.0.0.0")
APP_PORT = int(_env("CASE_PORT", "5050"))
RESUMABLE_THRESHOLD_BYTES = int(_env("CASE_RESUMABLE_THRESHOLD_MB", "100")) * 1024 * 1024
UPLOAD_CHUNK_SIZE_BYTES = int(_env("CASE_UPLOAD_CHUNK_MB", "8")) * 1024 * 1024
UPLOAD_TTL_HOURS = int(_env("CASE_UPLOAD_TTL_HOURS", "12"))
def get_unit_config(unit: str) -> UnitConfig:
key = unit.strip()
for name, cfg in UNITS.items():
if name.lower() == key.lower():
return cfg
raise ValueError(f"Unknown unit: {unit}. Allowed: {', '.join(ALLOWED_UNITS)}")