Add camera presets snapshots and continuous recording playback
This commit is contained in:
Codex
@@ -1,8 +1,10 @@
|
||||
"""Capability-driven ONVIF pan/tilt: bounded nudges, never latched movement."""
|
||||
import hashlib
|
||||
import json
|
||||
import threading
|
||||
import time
|
||||
import urllib.parse as url
|
||||
import xml.etree.ElementTree as ET
|
||||
from xml.sax.saxutils import escape
|
||||
import onvif
|
||||
|
||||
@@ -17,6 +19,100 @@ class Controller:
|
||||
self.guard = threading.RLock()
|
||||
self.cache = {}
|
||||
self.locks = {}
|
||||
self.cancellations = {}
|
||||
self.preset_versions = {}
|
||||
|
||||
def node_key(self, c, info):
|
||||
return c['host'], info['endpoint'][0], info['node']
|
||||
|
||||
def read_presets(self, c):
|
||||
info = self.discover(c)
|
||||
node = self.call(c, info['endpoint'], PTZ, 'GetNode',
|
||||
'<m:NodeToken>' + escape(info['node']) + '</m:NodeToken>')
|
||||
maximum = int(node.findtext('.//' + T + 'MaximumNumberOfPresets', '0'))
|
||||
if not 0 < maximum <= 1024:
|
||||
raise ValueError('设备未开放预置位功能')
|
||||
root = self.call(c, info['endpoint'], PTZ, 'GetPresets',
|
||||
'<m:ProfileToken>' + escape(info['profile']) + '</m:ProfileToken>')
|
||||
items = []
|
||||
for item in root.findall('.//{' + PTZ + '}Preset'):
|
||||
token = item.get('token', '')
|
||||
if not token or len(token) > 256 or any(p['token'] == token for p in items):
|
||||
raise ValueError('设备返回了无效预置位列表')
|
||||
items.append(dict(token=token, name=item.findtext(T + 'Name', '') or '未命名位置'))
|
||||
positions = [ET.tostring(p, encoding='unicode') for p in root.findall('.//{' + PTZ + '}Preset')]
|
||||
with self.guard:
|
||||
version = self.preset_versions.get(self.node_key(c, info), 0)
|
||||
fingerprint = [info['profile'], info['node'], maximum, sorted(items, key=lambda p: p['token']), sorted(positions), version]
|
||||
revision = hashlib.sha256(json.dumps(fingerprint, sort_keys=True).encode()).hexdigest()
|
||||
return dict(items=items, maximum=maximum, revision=revision,
|
||||
note='预置位保存在摄像机中，两个镜头可能共用云台和位置。'), info
|
||||
|
||||
def preset_command(self, c, data):
|
||||
action = data.get('action')
|
||||
if action not in ('create', 'remove', 'goto'):
|
||||
raise ValueError('请选择保存当前位置、删除或转到预置位')
|
||||
info = self.discover(c)
|
||||
key = self.node_key(c, info)
|
||||
with self.guard:
|
||||
lock = self.locks.setdefault(key, threading.Lock())
|
||||
if not lock.acquire(blocking=False):
|
||||
raise ValueError('云台正在执行操作，请稍后再点')
|
||||
try:
|
||||
snapshot, info = self.read_presets(c)
|
||||
if data.get('revision') != snapshot['revision']:
|
||||
raise ValueError('预置位已变化，请重新读取后操作')
|
||||
token = data.get('token')
|
||||
if action != 'create' and not any(p['token'] == token for p in snapshot['items']):
|
||||
raise ValueError('预置位不存在，请重新读取')
|
||||
body = '<m:ProfileToken>' + escape(info['profile']) + '</m:ProfileToken>'
|
||||
if action == 'goto':
|
||||
cancel = threading.Event()
|
||||
with self.guard:
|
||||
self.cancellations[key] = cancel
|
||||
try:
|
||||
# A finite preset move has no ONVIF Timeout parameter. Bound it on
|
||||
# the server and send Stop even after an ambiguous network failure.
|
||||
self.call(c, info['endpoint'], PTZ, 'GotoPreset', body + '<m:PresetToken>' + escape(token) + '</m:PresetToken>')
|
||||
deadline = time.monotonic() + 10
|
||||
while time.monotonic() < deadline:
|
||||
if cancel.wait(0.5):
|
||||
return dict(ok=True, completed=False, message='本次预置位转动已停止')
|
||||
status = self.call(c, info['endpoint'], PTZ, 'GetStatus', body)
|
||||
if status.findtext('.//' + T + 'MoveStatus/' + T + 'PanTilt') == 'IDLE':
|
||||
return dict(ok=True, completed=True, message='设备报告转动结束，请核对画面位置')
|
||||
return dict(ok=True, completed=False, message='设备未确认转动结束，已发送停止；请核对画面后重试')
|
||||
finally:
|
||||
try:
|
||||
self.call(c, info['endpoint'], PTZ, 'Stop', body + '<m:PanTilt>true</m:PanTilt><m:Zoom>true</m:Zoom>')
|
||||
finally:
|
||||
with self.guard:
|
||||
self.cancellations.pop(key, None)
|
||||
if action == 'create':
|
||||
name = data.get('name')
|
||||
if not isinstance(name, str) or not 1 <= len(name.strip()) <= 40 or any(ord(x) < 32 for x in name):
|
||||
raise ValueError('位置名称需要 1–40 个字且不包含控制字符')
|
||||
name = name.strip()
|
||||
if any(p['name'].casefold() == name.casefold() for p in snapshot['items']):
|
||||
raise ValueError('位置名称已存在，请换一个名称')
|
||||
if len(snapshot['items']) >= snapshot['maximum']:
|
||||
raise ValueError('设备预置位已满')
|
||||
payload = body + '<m:PresetName>' + escape(name) + '</m:PresetName>'
|
||||
with self.guard:
|
||||
self.preset_versions[key] = self.preset_versions.get(key, 0) + 1
|
||||
response = self.call(c, info['endpoint'], PTZ, 'SetPreset', payload)
|
||||
saved_token = response.findtext('.//{' + PTZ + '}PresetToken')
|
||||
else:
|
||||
with self.guard:
|
||||
self.preset_versions[key] = self.preset_versions.get(key, 0) + 1
|
||||
self.call(c, info['endpoint'], PTZ, 'RemovePreset', body + '<m:PresetToken>' + escape(token) + '</m:PresetToken>')
|
||||
latest, _ = self.read_presets(c)
|
||||
verified = (not any(p['token'] == token for p in latest['items']) if action == 'remove'
|
||||
else any(p['token'] == saved_token and p['name'] == name for p in latest['items']))
|
||||
return dict(ok=True, verified=verified, presets=latest,
|
||||
message='设备预置位列表已回读确认' if verified else '设备响应与预置位回读不一致，请核对后再操作')
|
||||
finally:
|
||||
lock.release()
|
||||
|
||||
def call(self, c, endpoint, namespace, operation, body=''):
|
||||
port, path = endpoint
|
||||
@@ -98,6 +194,10 @@ class Controller:
|
||||
def stop():
|
||||
self.call(c, info['endpoint'], PTZ, 'Stop', body + '<m:PanTilt>true</m:PanTilt>')
|
||||
if action == 'stop':
|
||||
with self.guard:
|
||||
cancel = self.cancellations.get(self.node_key(c, info))
|
||||
if cancel:
|
||||
cancel.set()
|
||||
stop()
|
||||
return {'ok': True, 'stopped': True}
|
||||
# Lock by physical host/node, including commands sent through another lens/account.
|
||||
|
||||
Reference in New Issue
Block a user