|
|
from __future__ import annotations
|
|
|
import logging
|
|
|
import xml.etree.ElementTree as ET
|
|
|
from typing import Tuple, Optional
|
|
|
|
|
|
import requests
|
|
|
from requests.auth import HTTPDigestAuth
|
|
|
|
|
|
from . import config
|
|
|
|
|
|
logger = logging.getLogger("PTZTracker")
|
|
|
|
|
|
# Универсальные таймауты и заголовки
|
|
|
_REQ_TIMEOUT = (1.2, 2.5) # (connect, read)
|
|
|
_XML = {"Content-Type": "application/xml; charset=UTF-8"}
|
|
|
|
|
|
def _auth(cam_id: int) -> Tuple[str, HTTPDigestAuth, int]:
|
|
|
cfg = config.CAMERA_CONFIG[cam_id]
|
|
|
ip = cfg["ip"]
|
|
|
user = cfg.get("username") or cfg.get("user") or "admin"
|
|
|
pw = cfg.get("password") or ""
|
|
|
ch = int(cfg.get("ptz_channel", 1))
|
|
|
return ip, HTTPDigestAuth(user, pw), ch
|
|
|
|
|
|
def _get_xml(url: str, auth: HTTPDigestAuth) -> Optional[ET.Element]:
|
|
|
try:
|
|
|
r = requests.get(url, auth=auth, timeout=_REQ_TIMEOUT)
|
|
|
r.raise_for_status()
|
|
|
return ET.fromstring(r.content)
|
|
|
except Exception as e:
|
|
|
logger.debug("[ISAPI] GET %s failed: %s", url, e)
|
|
|
return None
|
|
|
|
|
|
def _put_xml(url: str, auth: HTTPDigestAuth, xml_body: str) -> bool:
|
|
|
try:
|
|
|
r = requests.put(url, data=xml_body.encode("utf-8"), headers=_XML, auth=auth, timeout=_REQ_TIMEOUT)
|
|
|
r.raise_for_status()
|
|
|
return 200 <= r.status_code < 300
|
|
|
except Exception as e:
|
|
|
logger.debug("[ISAPI] PUT %s failed: %s", url, e)
|
|
|
return False
|
|
|
|
|
|
# ---------- Public API ----------
|
|
|
|
|
|
def get_ptz_status(cam_id: int) -> Tuple[Optional[float], Optional[float]]:
|
|
|
"""
|
|
|
Возвращает (pan_deg, tilt_deg) для камеры через ISAPI.
|
|
|
Хиквишн обычно отдаёт абсолютные значения pan/tilt в градусах.
|
|
|
Иногда встречается масштаб *10 (0..3600) — нормализуем.
|
|
|
"""
|
|
|
ip, auth, ch = _auth(cam_id)
|
|
|
url = f"http://{ip}/ISAPI/PTZCtrl/channels/{ch}/status"
|
|
|
root = _get_xml(url, auth)
|
|
|
if root is None:
|
|
|
return None, None
|
|
|
|
|
|
# Ищем разные варианты тегов
|
|
|
pan = None
|
|
|
tilt = None
|
|
|
for tag in ("pan", "absolutePan", "azimuth"):
|
|
|
el = root.find(f".//{tag}")
|
|
|
if el is not None and el.text:
|
|
|
try:
|
|
|
pan = float(el.text)
|
|
|
break
|
|
|
except Exception:
|
|
|
pass
|
|
|
for tag in ("tilt", "absoluteTilt", "elevation"):
|
|
|
el = root.find(f".//{tag}")
|
|
|
if el is not None and el.text:
|
|
|
try:
|
|
|
tilt = float(el.text)
|
|
|
break
|
|
|
except Exception:
|
|
|
pass
|
|
|
|
|
|
# Нормализация шкалы (редкие модели отдают *10)
|
|
|
def _norm(x: Optional[float]) -> Optional[float]:
|
|
|
if x is None:
|
|
|
return None
|
|
|
if abs(x) > 360.0 and abs(x) <= 3600.0:
|
|
|
x = x / 10.0
|
|
|
return float(x)
|
|
|
|
|
|
return _norm(pan), _norm(tilt)
|
|
|
|
|
|
def read_ptz_azimuth_deg(cam_id: int) -> Optional[float]:
|
|
|
pan, _ = get_ptz_status(cam_id)
|
|
|
return pan # pan — это и есть азимут головы в градусах
|
|
|
|
|
|
def goto_preset_token(cam_id: int, preset_token: str | int) -> bool:
|
|
|
"""
|
|
|
Переход к пресету по ISAPI. На HIK это:
|
|
|
PUT /ISAPI/PTZCtrl/channels/{ch}/presets/{id}/goto
|
|
|
<PTZPreset><id>{id}</id></PTZPreset>
|
|
|
"""
|
|
|
ip, auth, ch = _auth(cam_id)
|
|
|
pid = str(preset_token).strip()
|
|
|
url = f"http://{ip}/ISAPI/PTZCtrl/channels/{ch}/presets/{pid}/goto"
|
|
|
|
|
|
body = f"<PTZPreset><id>{pid}</id></PTZPreset>"
|
|
|
ok = _put_xml(url, auth, body)
|
|
|
if not ok:
|
|
|
logger.warning("[ISAPI] goto preset failed cam=%s ch=%s preset=%s", cam_id, ch, pid)
|
|
|
return ok
|