ucvl-home-vision/ptz.py

"""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', '<m:Category>All</m:Category>')
        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 = '<m:StreamSetup><tt:Stream>RTP-Unicast</tt:Stream><tt:Transport><tt:Protocol>RTSP</tt:Protocol></tt:Transport></m:StreamSetup><m:ProfileToken>' + escape(p.get('token', '')) + '</m:ProfileToken>'
                root = self.call(c, media, onvif.MEDIA, 'GetStreamUri', body)
                uri = root.find('.//' + T + 'Uri')
                actual, wanted_uri = url.urlsplit(uri.text if uri is not None else ''), url.urlsplit(c['mainPath'])
                if (actual.hostname == c['host'] and actual.path == wanted_uri.path
                        and sorted(url.parse_qsl(actual.query, keep_blank_values=True)) == sorted(url.parse_qsl(wanted_uri.query, keep_blank_values=True))):
                    selected = p
                    break
        if selected is None:
            raise ValueError('这个镜头没有匹配的云台配置')
        conf = selected.find(T + 'PTZConfiguration')
        options = self.call(c, endpoint, PTZ, 'GetConfigurationOptions', '<m:ConfigurationToken>' + escape(conf.get('token')) + '</m:ConfigurationToken>')
        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 = '<m:ProfileToken>' + escape(info['profile']) + '</m:ProfileToken>'
        def stop():
            self.call(c, info['endpoint'], PTZ, 'Stop', body + '<m:PanTilt>true</m:PanTilt>')
        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'<m:Velocity><tt:PanTilt x="{x}" y="{y}" space="{SPACE}"/></m:Velocity><m:Timeout>PT1S</m:Timeout>')
                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()