"""Capability-driven ONVIF pan/tilt: bounded nudges, never latched movement.""" import hashlib import threading import time import urllib.parse as url from xml.sax.saxutils import escape import onvif PTZ = 'http://www.onvif.org/ver20/ptz/wsdl' SPACE = 'http://www.onvif.org/ver10/tptz/PanTiltSpaces/VelocityGenericSpace' T = '{' + onvif.SCHEMA + '}' DIRECTIONS = {'left': (-0.25, 0), 'right': (0.25, 0), 'up': (0, 0.25), 'down': (0, -0.25)} class Controller: def __init__(self): self.guard = threading.RLock() self.cache = {} self.locks = {} def call(self, c, endpoint, namespace, operation, body=''): port, path = endpoint return onvif.request(c['host'], port, path, namespace, operation, body, c['username'], c['password']) @staticmethod def endpoint(c, text): parsed = url.urlsplit(text) if parsed.scheme != 'http' or parsed.hostname != c['host'] or parsed.username or parsed.password: raise ValueError('设备返回了不匹配的云台服务地址') return parsed.port or 80, parsed.path + ('?' + parsed.query if parsed.query else '') def discover(self, c): if c.get('sourceKind', 'direct') != 'direct': raise ValueError('请使用摄像头直连通道控制云台') key = (c['host'], c.get('onvifPort', 80), c['username'], hashlib.sha256(c['password'].encode()).hexdigest(), c['mainPath']) with self.guard: old = self.cache.get(key) if old and old[0] > time.monotonic(): return old[1] caps = self.call(c, (c.get('onvifPort', 80), '/onvif/device_service'), onvif.DEVICE, 'GetCapabilities', 'All') def service(name): value = caps.find('.//' + T + name + '/' + T + 'XAddr') if value is None or not value.text: raise ValueError('设备没有提供云台服务') return self.endpoint(c, value.text) endpoint, media = service('PTZ'), service('Media') profiles = self.call(c, media, onvif.MEDIA, 'GetProfiles') available = [p for p in profiles.findall('.//{' + onvif.MEDIA + '}Profiles') if p.find(T + 'PTZConfiguration') is not None] wanted = url.parse_qs(url.urlsplit(c['mainPath']).query).get('profile', [''])[0] selected = next((p for p in available if p.get('token') == wanted), None) if selected is None: for p in available[:32]: body = 'RTP-UnicastRTSP' + escape(p.get('token', '')) + '' root = self.call(c, media, onvif.MEDIA, 'GetStreamUri', body) uri = root.find('.//' + T + 'Uri') if uri is not None and url.urlsplit(uri.text).path == url.urlsplit(c['mainPath']).path: selected = p break if selected is None: raise ValueError('这个镜头没有匹配的云台配置') conf = selected.find(T + 'PTZConfiguration') options = self.call(c, endpoint, PTZ, 'GetConfigurationOptions', '' + escape(conf.get('token')) + '') spaces = options.findall('.//' + T + 'ContinuousPanTiltVelocitySpace') space = next((s for s in spaces if s.findtext(T + 'URI') == SPACE), None) if space is None: raise ValueError('设备不支持通用方向微调') for axis in ('XRange', 'YRange'): if float(space.findtext(T + axis + '/' + T + 'Min')) > -0.25 or float(space.findtext(T + axis + '/' + T + 'Max')) < 0.25: raise ValueError('设备云台速度范围不兼容') # Do not issue movement unless the device advertises a one-second watchdog. minimum = options.findtext('.//' + T + 'PTZTimeout/' + T + 'Min') maximum = options.findtext('.//' + T + 'PTZTimeout/' + T + 'Max') def seconds(value): import re m = re.fullmatch(r'PT(?:(\d+(?:\.\d+)?)H)?(?:(\d+(?:\.\d+)?)M)?(?:(\d+(?:\.\d+)?)S)?', value or '') return sum(float(v or 0) * scale for v, scale in zip(m.groups(), (3600, 60, 1))) if m else -1 if not 0 <= seconds(minimum) <= 1 <= seconds(maximum): raise ValueError('设备不支持一秒自动停止保护') result = {'endpoint': endpoint, 'profile': selected.get('token'), 'node': conf.findtext(T + 'NodeToken'), 'supported': True} with self.guard: if len(self.cache) > 128: self.cache.clear() self.cache[key] = (time.monotonic() + 300, result) return result def capabilities(self, c): self.discover(c) return {'supported': True, 'directions': list(DIRECTIONS), 'mode': 'nudge', 'note': '点击一次短距离转动并自动停止。同一摄像头的镜头可能共用云台。'} def command(self, c, action): if action not in (*DIRECTIONS, 'stop'): raise ValueError('请选择上、下、左、右或停止') info = self.discover(c) body = '' + escape(info['profile']) + '' def stop(): self.call(c, info['endpoint'], PTZ, 'Stop', body + 'true') if action == 'stop': stop() return {'ok': True, 'stopped': True} # Lock by physical host/node, including commands sent through another lens/account. key = (c['host'], info['endpoint'][0], info['node']) with self.guard: lock = self.locks.setdefault(key, threading.Lock()) if not lock.acquire(blocking=False): raise ValueError('云台正在执行操作,请稍后再点') try: x, y = DIRECTIONS[action] try: self.call(c, info['endpoint'], PTZ, 'ContinuousMove', body + f'PT1S') time.sleep(0.3) finally: # Also stop after an ambiguous network failure. Device timeout is a second guard. stop() return {'ok': True, 'stopped': True} finally: lock.release() CONTROLLER = Controller()