"""Contextra ONVIF discovery — RTSP stream URI resolution via ONVIF SOAP. Pure stdlib + `requests`. No ONVIF SDK, no GStreamer. Usage (stream manager): from app.onvif_discovery import resolve_onvif_stream, OnvifError res = resolve_onvif_stream("192.168.0.50", username="admin", password="...") # -> {"host": ..., "device_info": {...}, "profiles": [...], # "stream_uri": "rtsp://...", "profile_token": "Profile_1"} CLI: python3 onvif_discovery.py [--port 80] [--user U] [--pass P] [--profile TOKEN] """ import argparse import json import re import xml.etree.ElementTree as ET import requests MEDIA_NS = "http://www.onvif.org/ver10/media/wsdl" DEVICE_NS = "http://www.onvif.org/ver10/device/wsdl" SCHEMA_NS = "http://www.onvif.org/ver10/schema" SOAP_ENV_NS = "http://www.w3.org/2003/05/soap-envelope" class OnvifError(Exception): """ONVIF discovery failure (network, auth, SOAP fault, bad URI).""" def _localname(tag: str) -> str: return tag.rsplit("}", 1)[-1] if "}" in tag else tag def _findall(root, name: str): return [el for el in root.iter() if _localname(el.tag) == name] def _find(root, name: str): for el in root.iter(): if _localname(el.tag) == name: return el return None def _text(el) -> str: if el is None or el.text is None: return "" return el.text.strip() def _bool(el) -> bool: t = _text(el).lower() return t in ("true", "1") class OnvifClient: """Minimal ONVIF Media/Device client (SOAP 1.2 over HTTP, Basic auth).""" def __init__(self, host: str, port: int = 80, username: str | None = None, password: str | None = None, timeout: float = 5.0): self.username = username self.password = password self.timeout = float(timeout) h = (host or "").strip() m = re.match(r"^(?:https?://)?([A-Za-z0-9._\-]+)(?::(\d+))?$", h) if not m: raise OnvifError(f"cannot parse ONVIF host: {host!r}") self.host = m.group(1) self.port = int(m.group(2) or port) self.base_url = f"http://{self.host}:{self.port}" # -- SOAP plumbing ------------------------------------------------------ def _soap(self, action: str, body_xml: str, service_ns: str): """POST a SOAP envelope to /onvif/{Service}; return parsed response root. Raises OnvifError on HTTP error or SOAP Fault. """ url = f"{self.base_url}/onvif/{service_ns.rsplit('/', 1)[-1]}" prefix = service_ns.split("/")[-2] if "/" in service_ns else "t" envelope = ( '' f'' f"{body_xml}" ) headers = { "Content-Type": "application/soap+xml; charset=utf-8", "SOAPAction": f'"{service_ns}:{action}"', } auth = None if self.username is not None: auth = (self.username, self.password or "") try: resp = requests.post(url, data=envelope.encode("utf-8"), headers=headers, auth=auth, timeout=self.timeout) except requests.RequestException as e: raise OnvifError(f"ONVIF request to {url} failed: {e}") from e if resp.status_code == 401 or resp.status_code == 403: raise OnvifError( f"ONVIF auth rejected (HTTP {resp.status_code}) for " f"{self.host}:{self.port} — check username/password") if resp.status_code != 200: raise OnvifError(f"ONVIF HTTP {resp.status_code} from {url}") try: root = ET.fromstring(resp.content) except ET.ParseError as e: raise OnvifError(f"ONVIF response is not valid XML: {e}") from e fault = _find(root, "Fault") if fault is not None: code = _text(_find(fault, "Code")) or "?" reason = _text(_find(fault, "Reason")) or "" raise OnvifError(f"ONVIF SOAP Fault {code}: {reason}") return root # -- Device ------------------------------------------------------------- def get_device_info(self) -> dict: body = f'' root = self._soap("GetDeviceInformation", body, DEVICE_NS) r = _find(root, "GetDeviceInformationResponse") or root return { "manufacturer": _text(_find(r, "Manufacturer")), "model": _text(_find(r, "Model")), "firmware_version": _text(_find(r, "FirmwareVersion")), "serial_number": _text(_find(r, "SerialNumber")), } # -- Media -------------------------------------------------------------- def list_profiles(self) -> list[dict]: body = "" root = self._soap("GetProfiles", body, MEDIA_NS) r = _find(root, "GetProfilesResponse") or root profiles = [] for prof in _findall(r, "Profile"): # Token/Fixed: ONVIF 스펙은 자식 요소, 일부 디바이스는 속성 — 둘 다 지원 token_el = _find(prof, "Token") fixed_el = _find(prof, "Fixed") profiles.append({ "token": _text(token_el) or prof.get("Token", ""), "name": _text(_find(prof, "Name")) or prof.get("Name", ""), "fixed": _bool(fixed_el) if fixed_el is not None else (prof.get("Fixed", "").lower() in ("true", "1")), "has_video_source": _find(prof, "VideoSourceConfiguration") is not None, }) return profiles def get_stream_uri(self, profile_token: str | None = None, rtsp_only: bool = True) -> dict: if not profile_token: profiles = self.list_profiles() if not profiles: raise OnvifError("ONVIF device returned no profiles") # prefer first non-fixed profile with a video source config pick = next((p for p in profiles if not p["fixed"] and p["has_video_source"]), None) if pick is None: pick = next((p for p in profiles if not p["fixed"]), None) if pick is None: pick = profiles[0] # all fixed — best effort profile_token = pick["token"] body = ( f'' f'' f"RTSP" f"" f"{profile_token}" f"" ) root = self._soap("GetStreamUri", body, MEDIA_NS) r = _find(root, "GetStreamUriResponse") or root uri = _text(_find(r, "Uri")) valid_until = _text(_find(r, "ValidUntil")) if not uri: raise OnvifError(f"ONVIF GetStreamUri returned empty URI (profile={profile_token})") if rtsp_only and not uri.startswith("rtsp://"): raise OnvifError( f"ONVIF stream URI is not RTSP ({uri!r}) — " "device may only offer HTTP unicast; use source_type=rtsp with manual URI") return {"profile_token": profile_token, "stream_uri": uri, "valid_until": valid_until or None} def resolve_onvif_stream(host: str, port: int = 80, username: str | None = None, password: str | None = None, profile_token: str | None = None, timeout: float = 5.0) -> dict: """One-call convenience: device info + profiles + resolved RTSP URI.""" client = OnvifClient(host=host, port=port, username=username, password=password, timeout=timeout) try: device_info = client.get_device_info() profiles = client.list_profiles() uri_res = client.get_stream_uri(profile_token=profile_token) except OnvifError: raise return { "host": f"{client.host}:{client.port}", "device_info": device_info, "profiles": profiles, "stream_uri": uri_res["stream_uri"], "profile_token": uri_res["profile_token"], } if __name__ == "__main__": ap = argparse.ArgumentParser(description="ONVIF stream URI discovery") ap.add_argument("host", help="IP or host (optionally ip:port or http://ip)") ap.add_argument("--port", type=int, default=80) ap.add_argument("--user", default=None) ap.add_argument("--pass", dest="password", default=None) ap.add_argument("--profile", default=None, help="ProfileToken (default: auto-select)") args = ap.parse_args() try: res = resolve_onvif_stream(args.host, port=args.port, username=args.user, password=args.password, profile_token=args.profile) print(json.dumps(res, indent=2)) except OnvifError as e: print(f"ONVIF_ERROR: {e}") raise SystemExit(1)