226 lines
13 KiB
Python
226 lines
13 KiB
Python
|
"""Capability-driven, per-lens ONVIF settings. Never accepts an endpoint from a client."""
|
||
|
|
import copy
|
||
|
|
import hashlib
|
||
|
|
import json
|
||
|
|
import math
|
||
|
|
import threading
|
||
|
|
import urllib.parse as url
|
||
|
|
import xml.etree.ElementTree as ET
|
||
|
|
from xml.sax.saxutils import escape
|
||
|
|
import onvif
|
||
|
|
|
||
|
|
IMAGING = 'http://www.onvif.org/ver20/imaging/wsdl'
|
||
|
|
T = '{' + onvif.SCHEMA + '}'
|
||
|
|
M = '{' + onvif.MEDIA + '}'
|
||
|
|
I = '{' + IMAGING + '}'
|
||
|
|
IMAGE_FIELDS = {
|
||
|
|
'brightness': ('Brightness', 'Brightness', '亮度'),
|
||
|
|
'contrast': ('Contrast', 'Contrast', '对比度'),
|
||
|
|
'saturation': ('ColorSaturation', 'ColorSaturation', '饱和度'),
|
||
|
|
'sharpness': ('Sharpness', 'Sharpness', '锐度'),
|
||
|
|
'dayNight': ('IrCutFilter', 'IrCutFilterModes', '日夜模式'),
|
||
|
|
'backlight': ('BacklightCompensation/Mode', 'BacklightCompensation/Mode', '背光补偿'),
|
||
|
|
'wdr': ('WideDynamicRange/Mode', 'WideDynamicRange/Mode', '宽动态'),
|
||
|
|
'wdrLevel': ('WideDynamicRange/Level', 'WideDynamicRange/Level', '宽动态强度'),
|
||
|
|
'whiteBalance': ('WhiteBalance/Mode', 'WhiteBalance/Mode', '白平衡'),
|
||
|
|
}
|
||
|
|
ENUMS = {'dayNight': ('ON', 'OFF', 'AUTO'), 'backlight': ('ON', 'OFF'),
|
||
|
|
'wdr': ('ON', 'OFF'), 'whiteBalance': ('AUTO', 'MANUAL')}
|
||
|
|
VIDEO_FIELDS = {'fps': 'RateControl/FrameRateLimit', 'bitrate': 'RateControl/BitrateLimit',
|
||
|
|
'quality': 'Quality', 'gop': 'H264/GovLength'}
|
||
|
|
|
||
|
|
|
||
|
|
def path(value):
|
||
|
|
return '/'.join(T + part for part in value.split('/'))
|
||
|
|
|
||
|
|
|
||
|
|
def number(value):
|
||
|
|
try:
|
||
|
|
result = float(value)
|
||
|
|
return result if math.isfinite(result) else None
|
||
|
|
except (TypeError, ValueError):
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
def limits(root, name):
|
||
|
|
if root is None:
|
||
|
|
return None
|
||
|
|
minimum, maximum = number(root.findtext(path(name + '/Min'))), number(root.findtext(path(name + '/Max')))
|
||
|
|
return dict(min=minimum, max=maximum) if minimum is not None and maximum is not None and minimum <= maximum else None
|
||
|
|
|
||
|
|
|
||
|
|
def uri_path(value):
|
||
|
|
parsed = url.urlsplit(value)
|
||
|
|
return parsed.path, sorted(url.parse_qsl(parsed.query, keep_blank_values=True))
|
||
|
|
|
||
|
|
|
||
|
|
class Settings:
|
||
|
|
def __init__(self):
|
||
|
|
self.guard = threading.Lock()
|
||
|
|
self.locks = {}
|
||
|
|
|
||
|
|
def call(self, camera, endpoint, namespace, operation, body=''):
|
||
|
|
return onvif.request(camera['host'], endpoint[0], endpoint[1], namespace, operation,
|
||
|
|
body, camera['username'], camera['password'])
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def endpoint(camera, address):
|
||
|
|
parsed = url.urlsplit(address or '')
|
||
|
|
if parsed.scheme != 'http' or parsed.hostname != camera['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, camera, quality):
|
||
|
|
if camera.get('sourceKind', 'direct') != 'direct':
|
||
|
|
raise ValueError('摄像机参数需要直连摄像头；录像机通道请在录像机中配置')
|
||
|
|
if not camera.get('enabled'):
|
||
|
|
raise ValueError('请先启用这个镜头通道')
|
||
|
|
if quality not in ('main', 'sub'):
|
||
|
|
raise ValueError('请选择主码流或子码流')
|
||
|
|
device = (camera.get('onvifPort', 80), '/onvif/device_service')
|
||
|
|
caps = self.call(camera, device, onvif.DEVICE, 'GetCapabilities', '<m:Category>All</m:Category>')
|
||
|
|
services = {}
|
||
|
|
for name in ('Media', 'Imaging', 'PTZ', 'Events', 'Recording', 'Replay'):
|
||
|
|
address = caps.findtext('.//' + T + name + '/' + T + 'XAddr')
|
||
|
|
if address:
|
||
|
|
services[name] = self.endpoint(camera, address)
|
||
|
|
if 'Media' not in services:
|
||
|
|
raise ValueError('设备没有提供媒体配置服务')
|
||
|
|
profiles = self.call(camera, services['Media'], onvif.MEDIA, 'GetProfiles')
|
||
|
|
wanted = camera['mainPath' if quality == 'main' else 'subPath']
|
||
|
|
selected = None
|
||
|
|
# Exact path AND query matching prevents selecting the other lens on shared paths.
|
||
|
|
for profile in profiles.findall('.//' + M + 'Profiles')[:16]:
|
||
|
|
token = profile.get('token', '')
|
||
|
|
content = ('<m:StreamSetup><tt:Stream>RTP-Unicast</tt:Stream><tt:Transport><tt:Protocol>RTSP</tt:Protocol>'
|
||
|
|
'</tt:Transport></m:StreamSetup><m:ProfileToken>' + escape(token) + '</m:ProfileToken>')
|
||
|
|
response = self.call(camera, services['Media'], onvif.MEDIA, 'GetStreamUri', content)
|
||
|
|
uri = response.findtext('.//' + T + 'Uri')
|
||
|
|
if uri and url.urlsplit(uri).hostname == camera['host'] and uri_path(uri) == uri_path(wanted):
|
||
|
|
selected = profile
|
||
|
|
break
|
||
|
|
if selected is None:
|
||
|
|
raise ValueError('未找到与当前镜头及码流路径精确匹配的配置，请先核对通道')
|
||
|
|
source = selected.findtext(path('VideoSourceConfiguration/SourceToken'))
|
||
|
|
encoder = selected.find(T + 'VideoEncoderConfiguration')
|
||
|
|
if not source or encoder is None:
|
||
|
|
raise ValueError('设备没有返回镜头或编码器配置')
|
||
|
|
return dict(device=device, services=services, source=source, profile=selected.get('token'), encoder=encoder)
|
||
|
|
|
||
|
|
def read(self, camera, quality='main'):
|
||
|
|
info = self.discover(camera, quality)
|
||
|
|
return self.details(camera, info), info
|
||
|
|
|
||
|
|
def details(self, camera, info):
|
||
|
|
result = dict(profile=info['profile'], source=info['source'], device={}, imaging={}, video={}, notes=[],
|
||
|
|
advertisedServices=list(info['services']))
|
||
|
|
try:
|
||
|
|
device = self.call(camera, info['device'], onvif.DEVICE, 'GetDeviceInformation')
|
||
|
|
result['device'] = {key: device.findtext('.//{' + onvif.DEVICE + '}' + key, '')
|
||
|
|
for key in ('Manufacturer', 'Model', 'FirmwareVersion', 'HardwareId')}
|
||
|
|
except Exception:
|
||
|
|
result['notes'].append('设备型号读取失败，可重试')
|
||
|
|
encoder = info['encoder']
|
||
|
|
values = {key: number(encoder.findtext(path(field))) for key, field in VIDEO_FIELDS.items()}
|
||
|
|
values['resolution'] = encoder.findtext(path('Resolution/Width'), '') + 'x' + encoder.findtext(path('Resolution/Height'), '')
|
||
|
|
codec = encoder.get('encoding') or encoder.findtext(T + 'Encoding', '')
|
||
|
|
result['video'] = dict(values=values, options={}, codec=codec)
|
||
|
|
try:
|
||
|
|
body = '<m:ConfigurationToken>' + escape(encoder.get('token', '')) + '</m:ConfigurationToken>'
|
||
|
|
opts = self.call(camera, info['services']['Media'], onvif.MEDIA, 'GetVideoEncoderConfigurationOptions', body).find('.//' + M + 'Options')
|
||
|
|
# Some EZVIZ devices expose H265 through the H264 compatibility schema.
|
||
|
|
# Keep the entire original XML (including vendor encoding attribute) when writing.
|
||
|
|
schema_codec = encoder.findtext(T + 'Encoding', '')
|
||
|
|
if schema_codec not in ('H264', 'MPEG4', 'JPEG'):
|
||
|
|
raise ValueError('unsupported encoding schema')
|
||
|
|
codec_opts = opts.find(T + schema_codec)
|
||
|
|
for key, field in [('fps', 'FrameRateRange'), ('gop', 'GovLengthRange')]:
|
||
|
|
limit = limits(codec_opts, field)
|
||
|
|
if limit and values[key] is not None:result['video']['options'][key] = limit
|
||
|
|
bitrate = limits(opts, 'Extension/' + schema_codec + '/BitrateRange')
|
||
|
|
quality = limits(opts, 'QualityRange')
|
||
|
|
if bitrate and values['bitrate'] is not None:result['video']['options']['bitrate'] = bitrate
|
||
|
|
if quality and values['quality'] is not None:result['video']['options']['quality'] = quality
|
||
|
|
resolutions = [r.findtext(T + 'Width', '') + 'x' + r.findtext(T + 'Height', '')
|
||
|
|
for r in codec_opts.findall(T + 'ResolutionsAvailable')]
|
||
|
|
if resolutions:result['video']['options']['resolution'] = {'choices': list(dict.fromkeys(resolutions))}
|
||
|
|
except Exception:
|
||
|
|
result['notes'].append('码流可调范围读取失败，暂不可修改码流参数')
|
||
|
|
result['video']['options'] = {}
|
||
|
|
if 'Imaging' in info['services']:
|
||
|
|
try:
|
||
|
|
body = '<m:VideoSourceToken>' + escape(info['source']) + '</m:VideoSourceToken>'
|
||
|
|
settings = self.call(camera, info['services']['Imaging'], IMAGING, 'GetImagingSettings', body).find('.//' + I + 'ImagingSettings')
|
||
|
|
options = self.call(camera, info['services']['Imaging'], IMAGING, 'GetOptions', body).find('.//' + I + 'ImagingOptions')
|
||
|
|
if settings is None or options is None:raise ValueError('missing imaging response')
|
||
|
|
info['imagingXML'] = settings
|
||
|
|
for key, (field, option, label) in IMAGE_FIELDS.items():
|
||
|
|
value = settings.findtext(path(field))
|
||
|
|
if value is None:continue
|
||
|
|
if key in ENUMS:
|
||
|
|
choices = [v.text for v in options.findall(path(option)) if v.text in ENUMS[key]]
|
||
|
|
rule = {'choices': list(dict.fromkeys(choices))} if choices else None
|
||
|
|
else:
|
||
|
|
value, rule = number(value), limits(options, option)
|
||
|
|
if rule and value is not None:result['imaging'][key] = dict(value=value, label=label, **rule)
|
||
|
|
except Exception:
|
||
|
|
result['notes'].append('图像参数读取失败或设备未开放，暂不可修改图像参数')
|
||
|
|
else:
|
||
|
|
result['notes'].append('设备未开放图像调节服务')
|
||
|
|
fingerprint = dict(source=info['source'], profile=info['profile'], encoder=encoder.get('token'),
|
||
|
|
imaging=result['imaging'], video=result['video'])
|
||
|
|
result['revision'] = hashlib.sha256(json.dumps(fingerprint, sort_keys=True).encode()).hexdigest()
|
||
|
|
return result
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def validate(values, rules, integer_fields=()):
|
||
|
|
if not isinstance(values, dict) or not values or any(k not in rules for k in values):
|
||
|
|
raise ValueError('提交的参数未由设备开放，或没有需要保存的参数')
|
||
|
|
for key, value in values.items():
|
||
|
|
rule = rules[key]
|
||
|
|
if 'choices' in rule:
|
||
|
|
if not isinstance(value, str) or value not in rule['choices']:raise ValueError('参数选项不受设备支持')
|
||
|
|
elif (type(value) not in (int, float) or not math.isfinite(value) or not rule['min'] <= value <= rule['max']
|
||
|
|
or (key in integer_fields and int(value) != value)):
|
||
|
|
raise ValueError('参数超出设备允许范围')
|
||
|
|
|
||
|
|
def save(self, camera, data):
|
||
|
|
key = (camera['host'], camera.get('onvifPort', 80))
|
||
|
|
with self.guard:lock = self.locks.setdefault(key, threading.Lock())
|
||
|
|
if not lock.acquire(blocking=False):raise ValueError('这台摄像头正在保存参数，请稍后重试')
|
||
|
|
try:
|
||
|
|
current, info = self.read(camera, data.get('quality', 'main'))
|
||
|
|
if data.get('revision') != current['revision']:
|
||
|
|
raise ValueError('设备参数已变化，请重新读取后再修改')
|
||
|
|
section, values = data.get('section'), data.get('values')
|
||
|
|
if section == 'imaging':
|
||
|
|
self.validate(values, current['imaging'])
|
||
|
|
settings = copy.deepcopy(info['imagingXML'])
|
||
|
|
for key, value in values.items():settings.find(path(IMAGE_FIELDS[key][0])).text = str(value)
|
||
|
|
body = '<m:VideoSourceToken>' + escape(info['source']) + '</m:VideoSourceToken>' + ET.tostring(settings, encoding='unicode') + '<m:ForcePersistence>true</m:ForcePersistence>'
|
||
|
|
endpoint, namespace, operation = info['services']['Imaging'], IMAGING, 'SetImagingSettings'
|
||
|
|
elif section == 'video':
|
||
|
|
self.validate(values, current['video']['options'], ('fps', 'bitrate', 'gop'))
|
||
|
|
settings = copy.deepcopy(info['encoder']);settings.tag = M + 'Configuration'
|
||
|
|
for key, value in values.items():
|
||
|
|
if key == 'resolution':
|
||
|
|
for field, size in zip(('Width', 'Height'), value.split('x')):settings.find(path('Resolution/' + field)).text = size
|
||
|
|
else:settings.find(path(VIDEO_FIELDS[key])).text = str(int(value) if key in ('fps', 'bitrate', 'gop') else value)
|
||
|
|
body = ET.tostring(settings, encoding='unicode') + '<m:ForcePersistence>true</m:ForcePersistence>'
|
||
|
|
endpoint, namespace, operation = info['services']['Media'], onvif.MEDIA, 'SetVideoEncoderConfiguration'
|
||
|
|
else:raise ValueError('请选择图像或码流设置')
|
||
|
|
self.call(camera, endpoint, namespace, operation, body)
|
||
|
|
try:
|
||
|
|
fresh, _ = self.read(camera, data.get('quality', 'main'))
|
||
|
|
actual = {k: v['value'] for k, v in fresh['imaging'].items()} if section == 'imaging' else fresh['video']['values']
|
||
|
|
confirmed = all(actual.get(k) == v for k, v in values.items())
|
||
|
|
return dict(ok=True, verified=confirmed, settings=fresh,
|
||
|
|
message='设备已保存并回读确认' if confirmed else '设备已响应，但回读值与提交值不同，请核对设备实际参数')
|
||
|
|
except Exception:
|
||
|
|
return dict(ok=True, verified=False, settings=None, message='设备已响应，暂未读回参数；请重新读取确认，不要反复提交')
|
||
|
|
finally:
|
||
|
|
lock.release()
|
||
|
|
|
||
|
|
|
||
|
|
CONTROLLER = Settings()
|