Skip to content

Commit 78f6f38

Browse files
tamirOKtamirOK
authored
Support for custom device capabilities (martyanov#7)
* Support setting of custom capabilities * Handle missing XAddr
1 parent 69047ef commit 78f6f38

File tree

3 files changed

+231
-3
lines changed

3 files changed

+231
-3
lines changed

‎aonvif/client.py

+44-3
Original file line numberDiff line numberDiff line change
@@ -274,9 +274,47 @@ def __init__(
274274

275275
# Currently initialized services
276276
self._services = {}
277+
self._capabilities = None
277278

278279
self.to_dict = ONVIFService.to_dict
279280

281+
def set_capabilities(self, capabilities: dict):
282+
"""Set custom capabilities.
283+
284+
:param capabilities: Dictionary of camera capabilities.
285+
It must have the following format:
286+
{
287+
"Media": {
288+
"XAddr": "http://{host}/{xaddr_path_1},
289+
},
290+
"PTZ": {
291+
"XAddr": "http://{host}/{xaddr_path_2},
292+
},
293+
}
294+
"""
295+
296+
self._validate_capabilities(capabilities)
297+
self._capabilities = capabilities
298+
299+
def _validate_capabilities(self, capabilities: dict):
300+
if not isinstance(capabilities, dict):
301+
raise RuntimeError('Capabilities type must be dictionary')
302+
303+
for key, capability in capabilities.items():
304+
if not isinstance(key, str):
305+
raise RuntimeError('Capabilities key type must be string')
306+
307+
if not isinstance(capability, dict):
308+
raise RuntimeError('Capability type must be dictionary')
309+
310+
xaddr = capability.get('XAddr')
311+
312+
if xaddr is None:
313+
raise RuntimeError('Capability XAddr is missing')
314+
315+
if not isinstance(xaddr, str):
316+
raise RuntimeError('Capability XAddr type must be string')
317+
280318
async def update_xaddrs(self):
281319
"""Update xaddrs for services."""
282320

@@ -298,9 +336,12 @@ async def update_xaddrs(self):
298336

299337
# Get XAddr of services on the device
300338
self._xaddrs = {}
301-
capabilities = await devicemgmt.GetCapabilities({'Category': 'All'})
302-
for name in capabilities:
303-
capability = capabilities[name]
339+
340+
if not self._capabilities:
341+
self._capabilities = await devicemgmt.GetCapabilities({'Category': 'All'})
342+
343+
for name in self._capabilities:
344+
capability = self._capabilities[name]
304345
try:
305346
if name.lower() in wsdl.SERVICES and capability is not None:
306347
namespace = wsdl.SERVICES[name.lower()]['ns']

‎setup.py

+1
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,7 @@ def _get_long_description():
6666
'pytest-asyncio==0.20.2',
6767
'pytest-cov==4.0.0',
6868
'pytest==7.2.0',
69+
'pytest-mock==3.14.0',
6970
],
7071
},
7172
entry_points={

‎tests/test_client.py

+186
Original file line numberDiff line numberDiff line change
@@ -19,3 +19,189 @@ def maybe_raise(r=False):
1919
match='oops',
2020
):
2121
maybe_raise(True)
22+
23+
24+
def test_client_set_capabilies_with_invalid_capabilities_type():
25+
with pytest.raises(
26+
RuntimeError,
27+
match='Capabilities type must be dictionary',
28+
):
29+
client = aonvif.ONVIFCamera(
30+
'testhost',
31+
80,
32+
'username',
33+
'password',
34+
)
35+
client.set_capabilities(
36+
capabilities=[
37+
{
38+
'Media': {
39+
'XAddr': 'http://localhost/path',
40+
},
41+
},
42+
],
43+
)
44+
45+
46+
def test_client_set_capabilities_with_invalid_capabilities_key_type():
47+
with pytest.raises(
48+
RuntimeError,
49+
match='Capabilities key type must be string',
50+
):
51+
client = aonvif.ONVIFCamera(
52+
'testhost',
53+
80,
54+
'username',
55+
'password',
56+
)
57+
client.set_capabilities(
58+
capabilities={
59+
tuple('Media'): {
60+
'XAddr': 'http://localhost/path',
61+
},
62+
},
63+
)
64+
65+
66+
def test_client_set_capabilities_with_invalid_capability_type():
67+
with pytest.raises(
68+
RuntimeError,
69+
match='Capability type must be dictionary',
70+
):
71+
client = aonvif.ONVIFCamera(
72+
'testhost',
73+
80,
74+
'username',
75+
'password',
76+
)
77+
client.set_capabilities(
78+
capabilities={
79+
'Media': ['XAddr', 'http://localhost/path'],
80+
},
81+
)
82+
83+
84+
def test_client_set_capabilities_with_missing_xaddr():
85+
with pytest.raises(
86+
RuntimeError,
87+
match='Capability XAddr is missing',
88+
):
89+
client = aonvif.ONVIFCamera(
90+
'testhost',
91+
80,
92+
'username',
93+
'password',
94+
)
95+
client.set_capabilities(
96+
capabilities={
97+
'Media': {
98+
'RTPMulticast': True,
99+
},
100+
},
101+
)
102+
103+
104+
def test_client_set_capabilities_with_invalid_xaddr_type():
105+
with pytest.raises(
106+
RuntimeError,
107+
match='Capability XAddr type must be string',
108+
):
109+
client = aonvif.ONVIFCamera(
110+
'testhost',
111+
80,
112+
'username',
113+
'password',
114+
)
115+
client.set_capabilities(
116+
capabilities={
117+
'Media': {
118+
'XAddr': True,
119+
},
120+
},
121+
)
122+
123+
124+
@pytest.fixture
125+
def mocked_device_mgmt_service(mocker):
126+
mocked_service = mocker.AsyncMock()
127+
mocked_service.GetCapabilities.return_value = {
128+
'Analytics': {
129+
'XAddr': 'http://testhost/onvif/analytics_service',
130+
'RuleSupport': True,
131+
'AnalyticsModuleSupport': True,
132+
'_value_1': None,
133+
'_attr_1': None,
134+
},
135+
'Events': {
136+
'XAddr': 'http://testhost/onvif/event_service',
137+
'WSSubscriptionPolicySupport': True,
138+
'WSPullPointSupport': True,
139+
'WSPausableSubscriptionManagerInterfaceSupport': False,
140+
'_value_1': None,
141+
'_attr_1': None,
142+
},
143+
'PTZ': {
144+
'XAddr': 'http://testhost/onvif/ptz_service',
145+
'_value_1': None,
146+
'_attr_1': None,
147+
},
148+
}
149+
150+
return mocked_service
151+
152+
153+
@pytest.mark.asyncio
154+
async def test_client_update_xaddrs(mocker, mocked_device_mgmt_service):
155+
client = aonvif.ONVIFCamera(
156+
'testhost',
157+
80,
158+
'username',
159+
'password',
160+
)
161+
mocker.patch.object(
162+
client,
163+
'create_devicemgmt_service',
164+
return_value=mocked_device_mgmt_service,
165+
)
166+
167+
await client.update_xaddrs()
168+
169+
assert client._xaddrs == {
170+
'http://www.onvif.org/ver20/ptz/wsdl': 'http://testhost/onvif/ptz_service',
171+
'http://www.onvif.org/ver10/events/wsdl': 'http://testhost/onvif/event_service',
172+
'http://www.onvif.org/ver20/analytics/wsdl': 'http://testhost/onvif/analytics_service',
173+
}
174+
mocked_device_mgmt_service.GetCapabilities.assert_awaited_once_with({'Category': 'All'})
175+
176+
177+
@pytest.mark.asyncio
178+
async def test_client_update_xaddrs_with_custom_capabilities(
179+
mocker,
180+
mocked_device_mgmt_service,
181+
):
182+
client = aonvif.ONVIFCamera(
183+
'testhost',
184+
80,
185+
'username',
186+
'password',
187+
)
188+
client.set_capabilities(
189+
capabilities={
190+
'Imaging': {
191+
'XAddr': 'http://testhost/onvif/imaging_service',
192+
},
193+
},
194+
)
195+
196+
mocker.patch.object(
197+
client,
198+
'create_devicemgmt_service',
199+
return_value=mocked_device_mgmt_service,
200+
)
201+
202+
await client.update_xaddrs()
203+
204+
assert client._xaddrs == {
205+
'http://www.onvif.org/ver20/imaging/wsdl': 'http://testhost/onvif/imaging_service',
206+
}
207+
mocked_device_mgmt_service.GetCapabilities.assert_not_awaited()

0 commit comments

Comments
 (0)