ilaandClaude Opus 4.8
d3cd8ecb55
Adds write_local_event_tuning and a sensitivity group (confirm window, horizontal angle, require-rapid-drop, require-lower-body) that persists to the untracked local config and applies next start. 79 tests pass. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
340 lines
12 KiB
Python
340 lines
12 KiB
Python
"""Validated, credential-safe configuration loading for Silver Pose V1."""
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any, Dict
|
|
from urllib.parse import quote
|
|
|
|
|
|
class ConfigError(ValueError):
|
|
"""Raised when a configuration file cannot safely start V1."""
|
|
|
|
|
|
# Hikvision RTSP path: channel "101" = channel 1 main stream, "102" = sub stream.
|
|
_HIK_PATH = "/Streaming/Channels/{0}"
|
|
|
|
|
|
def build_rtsp_url(
|
|
host: str,
|
|
port: int,
|
|
username: str,
|
|
password: str,
|
|
channel: str = "101",
|
|
path_template: str = _HIK_PATH,
|
|
) -> str:
|
|
"""Build an RTSP URL, percent-encoding credentials per RFC 3986.
|
|
|
|
Encoding the username and password means special characters such as ``@``,
|
|
``:`` and ``/`` in a password no longer break URL parsing. The returned URL
|
|
holds credentials and must never be logged or persisted.
|
|
"""
|
|
|
|
user = quote(str(username), safe="")
|
|
secret = quote(str(password), safe="")
|
|
path = path_template.format(quote(str(channel), safe=""))
|
|
return "rtsp://{0}:{1}@{2}:{3}{4}".format(user, secret, host, int(port), path)
|
|
|
|
|
|
def build_ffmpeg_options(
|
|
transport: str = "tcp",
|
|
timeout_seconds: float = 5.0,
|
|
low_latency: bool = True,
|
|
) -> str:
|
|
"""Build the OpenCV FFmpeg capture options string from friendly settings.
|
|
|
|
OpenCV reads ``OPENCV_FFMPEG_CAPTURE_OPTIONS`` when the stream is opened, so
|
|
the caller must set this into the environment before ``cv2.VideoCapture``.
|
|
Format is ``key;value`` pairs joined by ``|``.
|
|
"""
|
|
|
|
if transport not in ("tcp", "udp"):
|
|
raise ValueError("transport must be 'tcp' or 'udp'")
|
|
parts = ["rtsp_transport;{0}".format(transport)]
|
|
if timeout_seconds and float(timeout_seconds) > 0:
|
|
parts.append("stimeout;{0}".format(int(float(timeout_seconds) * 1_000_000)))
|
|
if low_latency:
|
|
parts.append("fflags;nobuffer")
|
|
parts.append("flags;low_delay")
|
|
return "|".join(parts)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class EventConfig:
|
|
keypoint_confidence_threshold: float
|
|
suspect_window_seconds: float
|
|
confirm_window_seconds: float
|
|
recovery_window_seconds: float
|
|
cooldown_seconds: float
|
|
require_rapid_drop: bool = False
|
|
require_lower_body: bool = False
|
|
horizontal_angle_threshold_degrees: float = 45.0
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AppConfig:
|
|
source_id: str
|
|
source_url: str
|
|
model_path: Path
|
|
model_sha256: str
|
|
confidence_threshold: float
|
|
event: EventConfig
|
|
event_dir: Path
|
|
source_mode: str = "stream"
|
|
transport: str = "tcp"
|
|
timeout_seconds: float = 5.0
|
|
low_latency: bool = True
|
|
|
|
@property
|
|
def ffmpeg_capture_options(self) -> str:
|
|
"""OpenCV FFmpeg options string to set before opening the stream."""
|
|
|
|
return build_ffmpeg_options(self.transport, self.timeout_seconds, self.low_latency)
|
|
|
|
@property
|
|
def runtime_config_version(self) -> str:
|
|
"""Return a stable, non-secret identifier for the active event settings."""
|
|
|
|
payload = {
|
|
"source_id": self.source_id,
|
|
"model_sha256": self.model_sha256,
|
|
"confidence_threshold": self.confidence_threshold,
|
|
"event": {
|
|
"keypoint_confidence_threshold": self.event.keypoint_confidence_threshold,
|
|
"suspect_window_seconds": self.event.suspect_window_seconds,
|
|
"confirm_window_seconds": self.event.confirm_window_seconds,
|
|
"recovery_window_seconds": self.event.recovery_window_seconds,
|
|
"cooldown_seconds": self.event.cooldown_seconds,
|
|
"require_rapid_drop": self.event.require_rapid_drop,
|
|
"require_lower_body": self.event.require_lower_body,
|
|
"horizontal_angle_threshold_degrees": self.event.horizontal_angle_threshold_degrees,
|
|
},
|
|
}
|
|
canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
|
|
return "cfg-" + hashlib.sha256(canonical.encode("utf-8")).hexdigest()
|
|
|
|
|
|
_ENVIRONMENT_NAME = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
|
|
_SHA256 = re.compile(r"^[0-9a-fA-F]{64}$")
|
|
|
|
|
|
def _mapping(value: Any, field_name: str) -> Dict[str, Any]:
|
|
if not isinstance(value, dict):
|
|
raise ConfigError("{0} must be an object".format(field_name))
|
|
return value
|
|
|
|
|
|
def _text(value: Any, field_name: str) -> str:
|
|
if not isinstance(value, str) or not value.strip():
|
|
raise ConfigError("{0} must be a non-empty string".format(field_name))
|
|
return value.strip()
|
|
|
|
|
|
def _number(value: Any, field_name: str, minimum: float, maximum: float) -> float:
|
|
try:
|
|
result = float(value)
|
|
except (TypeError, ValueError):
|
|
raise ConfigError("{0} must be numeric".format(field_name))
|
|
if not minimum <= result <= maximum:
|
|
raise ConfigError(
|
|
"{0} must be between {1} and {2}".format(field_name, minimum, maximum)
|
|
)
|
|
return result
|
|
|
|
|
|
def _bool(value: Any, field_name: str) -> bool:
|
|
if not isinstance(value, bool):
|
|
raise ConfigError("{0} must be a boolean".format(field_name))
|
|
return value
|
|
|
|
|
|
def _resolve_path(config_path: Path, value: Any, field_name: str) -> Path:
|
|
raw_path = Path(_text(value, field_name))
|
|
if raw_path.is_absolute():
|
|
return raw_path
|
|
return (config_path.parent / raw_path).resolve()
|
|
|
|
|
|
def write_local_camera_source(
|
|
config_path: Path,
|
|
source_id: str,
|
|
host: str,
|
|
port: int,
|
|
channel: str,
|
|
username: str,
|
|
password: str,
|
|
mode: str = "stream",
|
|
transport: str = "tcp",
|
|
timeout_seconds: float = 5.0,
|
|
low_latency: bool = True,
|
|
) -> Path:
|
|
"""Persist a structured camera source into an untracked local config file.
|
|
|
|
Only ever call this on ``config.local.json`` (git-ignored). Credentials are
|
|
written in plaintext by design of the chosen policy; the public example must
|
|
never receive them.
|
|
"""
|
|
|
|
path = Path(config_path)
|
|
data = json.loads(path.read_text(encoding="utf-8")) if path.exists() else {}
|
|
data["source"] = {
|
|
"id": str(source_id),
|
|
"host": str(host),
|
|
"port": int(port),
|
|
"channel": str(channel),
|
|
"username": str(username),
|
|
"password": str(password),
|
|
"mode": mode,
|
|
"transport": transport,
|
|
"timeout_seconds": float(timeout_seconds),
|
|
"low_latency": bool(low_latency),
|
|
}
|
|
path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
return path
|
|
|
|
|
|
def write_local_event_tuning(
|
|
config_path: Path,
|
|
require_rapid_drop: bool,
|
|
require_lower_body: bool,
|
|
horizontal_angle_threshold_degrees: float,
|
|
confirm_window_seconds: float,
|
|
) -> Path:
|
|
"""Persist detection-sensitivity tuning into the local config's event block.
|
|
|
|
Only the tuning keys are updated; other event fields are preserved. Applies
|
|
at the next start (the running config is a start-time snapshot).
|
|
"""
|
|
|
|
path = Path(config_path)
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
event = dict(data.get("event", {}))
|
|
event["require_rapid_drop"] = bool(require_rapid_drop)
|
|
event["require_lower_body"] = bool(require_lower_body)
|
|
event["horizontal_angle_threshold_degrees"] = float(horizontal_angle_threshold_degrees)
|
|
event["confirm_window_seconds"] = float(confirm_window_seconds)
|
|
data["event"] = event
|
|
path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
return path
|
|
|
|
|
|
def load_config(path: Path) -> AppConfig:
|
|
"""Load one public/local configuration pair without persisting credentials.
|
|
|
|
The config file can name an environment variable but must not embed a source
|
|
address. The resolved address remains in memory only.
|
|
"""
|
|
|
|
config_path = Path(path).resolve()
|
|
try:
|
|
raw = json.loads(config_path.read_text(encoding="utf-8"))
|
|
except OSError as exc:
|
|
raise ConfigError("cannot read config: {0}".format(exc))
|
|
except json.JSONDecodeError as exc:
|
|
raise ConfigError("invalid JSON config: {0}".format(exc))
|
|
|
|
root = _mapping(raw, "config")
|
|
source = _mapping(root.get("source"), "source")
|
|
if "url" in source or "rtsp_url" in source:
|
|
raise ConfigError("source must define structured host fields or rtsp_url_env, not a full URL")
|
|
|
|
if "host" in source:
|
|
host = _text(source.get("host"), "source.host")
|
|
port = int(_number(source.get("port", 554), "source.port", 1.0, 65535.0))
|
|
channel = str(source.get("channel", "101")).strip()
|
|
if not channel:
|
|
raise ConfigError("source.channel must be a non-empty value")
|
|
username = _text(source.get("username"), "source.username")
|
|
password = source.get("password")
|
|
if not isinstance(password, str) or password == "":
|
|
raise ConfigError("source.password must be a non-empty string")
|
|
source_url = build_rtsp_url(host, port, username, password, channel)
|
|
else:
|
|
environment_name = _text(source.get("rtsp_url_env"), "source.rtsp_url_env")
|
|
if not _ENVIRONMENT_NAME.match(environment_name):
|
|
raise ConfigError("source.rtsp_url_env must be an environment variable name")
|
|
source_url = os.environ.get(environment_name)
|
|
if not source_url:
|
|
raise ConfigError("missing RTSP environment variable: {0}".format(environment_name))
|
|
|
|
source_mode = source.get("mode", "stream")
|
|
if source_mode not in ("replay", "stream"):
|
|
raise ConfigError("source.mode must be 'replay' or 'stream'")
|
|
|
|
transport = str(source.get("transport", "tcp")).strip().lower()
|
|
if transport not in ("tcp", "udp"):
|
|
raise ConfigError("source.transport must be 'tcp' or 'udp'")
|
|
timeout_seconds = _number(
|
|
source.get("timeout_seconds", 5.0), "source.timeout_seconds", 0.0, 60.0
|
|
)
|
|
low_latency = source.get("low_latency", True)
|
|
if not isinstance(low_latency, bool):
|
|
raise ConfigError("source.low_latency must be a boolean")
|
|
|
|
model = _mapping(root.get("model"), "model")
|
|
model_sha256 = _text(model.get("sha256"), "model.sha256").lower()
|
|
if not _SHA256.match(model_sha256):
|
|
raise ConfigError("model.sha256 must be a 64-character SHA-256 value")
|
|
|
|
event_raw = _mapping(root.get("event"), "event")
|
|
event = EventConfig(
|
|
keypoint_confidence_threshold=_number(
|
|
event_raw.get("keypoint_confidence_threshold"),
|
|
"event.keypoint_confidence_threshold",
|
|
0.0,
|
|
1.0,
|
|
),
|
|
suspect_window_seconds=_number(
|
|
event_raw.get("suspect_window_seconds"),
|
|
"event.suspect_window_seconds",
|
|
0.0,
|
|
30.0,
|
|
),
|
|
confirm_window_seconds=_number(
|
|
event_raw.get("confirm_window_seconds"),
|
|
"event.confirm_window_seconds",
|
|
1.0,
|
|
3.0,
|
|
),
|
|
recovery_window_seconds=_number(
|
|
event_raw.get("recovery_window_seconds"),
|
|
"event.recovery_window_seconds",
|
|
0.0,
|
|
300.0,
|
|
),
|
|
cooldown_seconds=_number(
|
|
event_raw.get("cooldown_seconds"),
|
|
"event.cooldown_seconds",
|
|
0.0,
|
|
3600.0,
|
|
),
|
|
require_rapid_drop=_bool(event_raw.get("require_rapid_drop", False), "event.require_rapid_drop"),
|
|
require_lower_body=_bool(event_raw.get("require_lower_body", False), "event.require_lower_body"),
|
|
horizontal_angle_threshold_degrees=_number(
|
|
event_raw.get("horizontal_angle_threshold_degrees", 45.0),
|
|
"event.horizontal_angle_threshold_degrees",
|
|
0.0,
|
|
90.0,
|
|
),
|
|
)
|
|
|
|
artifacts = _mapping(root.get("artifacts"), "artifacts")
|
|
return AppConfig(
|
|
source_id=_text(source.get("id"), "source.id"),
|
|
source_url=source_url,
|
|
model_path=_resolve_path(config_path, model.get("path"), "model.path"),
|
|
model_sha256=model_sha256,
|
|
confidence_threshold=_number(
|
|
model.get("confidence_threshold"), "model.confidence_threshold", 0.0, 1.0
|
|
),
|
|
event=event,
|
|
event_dir=_resolve_path(config_path, artifacts.get("event_dir"), "artifacts.event_dir"),
|
|
source_mode=source_mode,
|
|
transport=transport,
|
|
timeout_seconds=timeout_seconds,
|
|
low_latency=low_latency,
|
|
)
|