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 {id} """ ip, auth, ch = _auth(cam_id) pid = str(preset_token).strip() url = f"http://{ip}/ISAPI/PTZCtrl/channels/{ch}/presets/{pid}/goto" body = f"{pid}" 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