Skip to content

Commit b278de0

Browse files
kaburagiseckaburagisec
committed
feat(error-handling): Add ONVIFErrorHandler for graceful error management in ONVIF operations
1 parent f8c88b2 commit b278de0

File tree

5 files changed

+338
-3
lines changed

5 files changed

+338
-3
lines changed

‎examples/error_handling.py‎

Lines changed: 245 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,245 @@
1+
"""
2+
Path: examples/error_handling.py
3+
Author: @kaburagisec
4+
Created: October 14, 2025
5+
Tested devices: TP-Link Tapo C210 (https://www.tp-link.com/en/home-networking/cloud-camera/tapo-c210/)
6+
7+
This example demonstrates various ways to handle ONVIF operation errors,
8+
especially dealing with ActionNotSupported errors from devices that don't
9+
support certain features.
10+
11+
Applied for (>= v0.0.7 patch)
12+
"""
13+
14+
from onvif import ONVIFClient, CacheMode, ONVIFErrorHandler, ONVIFOperationException
15+
16+
HOST = "192.168.1.14"
17+
PORT = 2020
18+
USERNAME = "admintapo"
19+
PASSWORD = "admin123"
20+
21+
22+
def example_1_safe_call():
23+
"""
24+
Example 1: Using safe_call utility
25+
26+
safe_call automatically handles ActionNotSupported errors and returns
27+
a default value instead of raising an exception.
28+
"""
29+
print("=" * 60)
30+
print("Example 1: Using safe_call")
31+
print("=" * 60)
32+
33+
client = ONVIFClient(HOST, PORT, USERNAME, PASSWORD, cache=CacheMode.NONE)
34+
device = client.devicemgmt()
35+
36+
# Returns None if operation is not supported
37+
ip_filter = ONVIFErrorHandler.safe_call(lambda: device.GetIPAddressFilter())
38+
if ip_filter:
39+
print(f"✓ IP Address Filter: {ip_filter}")
40+
else:
41+
print("⚠ GetIPAddressFilter not supported or returned None")
42+
43+
# Returns empty list if operation is not supported
44+
dns_info = ONVIFErrorHandler.safe_call(
45+
lambda: device.GetDNS(),
46+
default={"FromDHCP": False, "DNSManual": []}
47+
)
48+
print(f"✓ DNS Info: {dns_info}")
49+
50+
print()
51+
52+
53+
def example_2_decorator():
54+
"""
55+
Example 2: Using @ignore_unsupported decorator
56+
57+
The decorator wraps a function to automatically handle ActionNotSupported
58+
errors and return None instead.
59+
"""
60+
print("=" * 60)
61+
print("Example 2: Using @ignore_unsupported decorator")
62+
print("=" * 60)
63+
64+
client = ONVIFClient(HOST, PORT, USERNAME, PASSWORD, cache=CacheMode.NONE)
65+
device = client.devicemgmt()
66+
67+
@ONVIFErrorHandler.ignore_unsupported
68+
def get_zero_configuration():
69+
return device.GetZeroConfiguration()
70+
71+
@ONVIFErrorHandler.ignore_unsupported
72+
def get_ntp():
73+
return device.GetNTP()
74+
75+
zero_conf = get_zero_configuration()
76+
if zero_conf:
77+
print(f"✓ Zero Configuration: {zero_conf}")
78+
else:
79+
print("⚠ GetZeroConfiguration not supported")
80+
81+
ntp = get_ntp()
82+
if ntp:
83+
print(f"✓ NTP: {ntp}")
84+
else:
85+
print("⚠ GetNTP not supported")
86+
87+
print()
88+
89+
90+
def example_3_manual_handling():
91+
"""
92+
Example 3: Manual exception handling
93+
94+
For more fine-grained control, catch ONVIFOperationException and use
95+
is_action_not_supported() to check the error type.
96+
"""
97+
print("=" * 60)
98+
print("Example 3: Manual exception handling")
99+
print("=" * 60)
100+
101+
client = ONVIFClient(HOST, PORT, USERNAME, PASSWORD, cache=CacheMode.NONE)
102+
device = client.devicemgmt()
103+
104+
# Try GetSystemUris (not always supported), fallback to alternative
105+
try:
106+
system_uris = device.GetSystemUris()
107+
print(f"✓ System URIs:")
108+
109+
# System Log URIs (can be multiple)
110+
if hasattr(system_uris, 'SystemLogUris') and system_uris.SystemLogUris:
111+
log_uris = system_uris.SystemLogUris
112+
if hasattr(log_uris, 'SystemLog'):
113+
logs = log_uris.SystemLog if isinstance(log_uris.SystemLog, list) else [log_uris.SystemLog]
114+
for log in logs:
115+
log_type = getattr(log, 'Type', 'Unknown')
116+
log_uri = getattr(log, 'Uri', 'N/A')
117+
print(f" System Log ({log_type}): {log_uri}")
118+
119+
# Support Info URI
120+
if hasattr(system_uris, 'SupportInfoUri') and system_uris.SupportInfoUri:
121+
print(f" Support Info: {system_uris.SupportInfoUri}")
122+
123+
# System Backup URI
124+
if hasattr(system_uris, 'SystemBackupUri') and system_uris.SystemBackupUri:
125+
print(f" System Backup: {system_uris.SystemBackupUri}")
126+
127+
except ONVIFOperationException as e:
128+
if ONVIFErrorHandler.is_action_not_supported(e):
129+
print("⚠ GetSystemUris not supported by this device")
130+
print(" Using alternative method to get device info...")
131+
132+
# Fallback: Get basic device information
133+
device_info = device.GetDeviceInformation()
134+
print(f"✓ Device Information (alternative):")
135+
print(f" Manufacturer: {getattr(device_info, 'Manufacturer', 'N/A')}")
136+
print(f" Model: {getattr(device_info, 'Model', 'N/A')}")
137+
print(f" FirmwareVersion: {getattr(device_info, 'FirmwareVersion', 'N/A')}")
138+
else:
139+
# Other errors should be re-raised
140+
print(f"✗ Unexpected error: {e}")
141+
raise
142+
143+
print()
144+
145+
146+
def example_4_batch_operations():
147+
"""
148+
Example 4: Batch operations with graceful degradation
149+
150+
Query multiple device features, collecting what's available and
151+
gracefully handling unsupported operations.
152+
"""
153+
print("=" * 60)
154+
print("Example 4: Batch operations with graceful degradation")
155+
print("=" * 60)
156+
157+
client = ONVIFClient(HOST, PORT, USERNAME, PASSWORD, cache=CacheMode.NONE)
158+
device = client.devicemgmt()
159+
160+
# Define operations to try
161+
operations = {
162+
"Device Info": lambda: device.GetDeviceInformation(),
163+
"System Date/Time": lambda: device.GetSystemDateAndTime(),
164+
"Network Interfaces": lambda: device.GetNetworkInterfaces(),
165+
"Hostname": lambda: device.GetHostname(),
166+
"DNS": lambda: device.GetDNS(),
167+
"NTP": lambda: device.GetNTP(),
168+
"Network Protocols": lambda: device.GetNetworkProtocols(),
169+
"IP Address Filter": lambda: device.GetIPAddressFilter(),
170+
"Zero Configuration": lambda: device.GetZeroConfiguration(),
171+
"Services": lambda: device.GetServices(IncludeCapability=False),
172+
}
173+
174+
results = {}
175+
supported_count = 0
176+
unsupported_count = 0
177+
178+
for name, operation in operations.items():
179+
result = ONVIFErrorHandler.safe_call(operation, default=None, log_error=False)
180+
results[name] = result
181+
182+
if result is not None:
183+
print(f"✓ {name}: Available")
184+
supported_count += 1
185+
else:
186+
print(f"⚠ {name}: Not supported")
187+
unsupported_count += 1
188+
189+
print(f"\nSummary: {supported_count} supported, {unsupported_count} not supported")
190+
print()
191+
192+
193+
def example_5_critical_operations():
194+
"""
195+
Example 5: Critical operations that should not be ignored
196+
197+
Some operations are critical and should fail loudly if not supported.
198+
Use ignore_unsupported=False for these.
199+
"""
200+
print("=" * 60)
201+
print("Example 5: Critical operations (don't ignore errors)")
202+
print("=" * 60)
203+
204+
client = ONVIFClient(HOST, PORT, USERNAME, PASSWORD, cache=CacheMode.NONE)
205+
device = client.devicemgmt()
206+
207+
# Critical operation - must succeed
208+
try:
209+
device_info = ONVIFErrorHandler.safe_call(
210+
lambda: device.GetDeviceInformation(),
211+
ignore_unsupported=False # Raise exception if not supported
212+
)
213+
print(f"✓ Device Info (critical):")
214+
print(f" Manufacturer: {getattr(device_info, 'Manufacturer', 'N/A')}")
215+
print(f" Model: {getattr(device_info, 'Model', 'N/A')}")
216+
print(f" FirmwareVersion: {getattr(device_info, 'FirmwareVersion', 'N/A')}")
217+
except ONVIFOperationException as e:
218+
print(f"✗ Critical operation failed: {e}")
219+
print(" Cannot continue without device information!")
220+
return
221+
222+
print()
223+
224+
225+
def main():
226+
"""Run all examples"""
227+
try:
228+
example_1_safe_call()
229+
example_2_decorator()
230+
example_3_manual_handling()
231+
example_4_batch_operations()
232+
example_5_critical_operations()
233+
234+
print("=" * 60)
235+
print("All examples completed successfully!")
236+
print("=" * 60)
237+
238+
except Exception as e:
239+
print(f"\n✗ Fatal error: {e}")
240+
import traceback
241+
traceback.print_exc()
242+
243+
244+
if __name__ == "__main__":
245+
main()

‎onvif/__init__.py‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22

33
from .client import ONVIFClient
44
from .operator import ONVIFOperator, CacheMode
5-
from .utils import ONVIFWSDL, ONVIFOperationException
5+
from .utils import ONVIFWSDL, ONVIFOperationException, ONVIFErrorHandler
66
from .utils.zeep import apply_patch, remove_patch, is_patched
77

88
__all__ = [
@@ -11,6 +11,7 @@
1111
"CacheMode",
1212
"ONVIFWSDL",
1313
"ONVIFOperationException",
14+
"ONVIFErrorHandler",
1415
"apply_patch",
1516
"remove_patch",
1617
"is_patched",

‎onvif/operator.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -117,7 +117,7 @@ def __init__(
117117
raise ValueError("Bindings must be set according to the WSDL service")
118118

119119
self.service = self.client.create_service(binding, self.address)
120-
logging.debug(f"ONVIFCore initialized {binding} at {self.address}")
120+
#logging.debug(f"ONVIFOperator initialized {binding} at {self.address}")
121121

122122
def call(self, method: str, *args, **kwargs):
123123
try:

‎onvif/utils/__init__.py‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,11 @@
11
from .wsdl import ONVIFWSDL
22
from .exceptions import ONVIFOperationException
3+
from .error_handlers import ONVIFErrorHandler
34
from .xml_capture import XMLCapturePlugin
45

5-
__all__ = ["ONVIFWSDL", "ONVIFOperationException", "XMLCapturePlugin"]
6+
__all__ = [
7+
"ONVIFWSDL",
8+
"ONVIFOperationException",
9+
"ONVIFErrorHandler",
10+
"XMLCapturePlugin",
11+
]

‎onvif/utils/error_handlers.py‎

Lines changed: 83 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,83 @@
1+
# onvif/utils/error_handlers.py
2+
3+
"""
4+
Error Handling Utilities for ONVIF Operations
5+
6+
Provides utilities to gracefully handle ONVIF operation errors,
7+
especially ActionNotSupported SOAP faults from devices that don't
8+
support certain operations.
9+
"""
10+
11+
import logging
12+
from zeep.exceptions import Fault
13+
from .exceptions import ONVIFOperationException
14+
15+
16+
class ONVIFErrorHandler:
17+
"""
18+
Error handling utilities for ONVIF operations.
19+
20+
Provides static methods to handle ONVIF errors gracefully,
21+
especially ActionNotSupported SOAP faults.
22+
"""
23+
24+
@staticmethod
25+
def is_action_not_supported(exception):
26+
"""Check if an ONVIFOperationException is caused by ActionNotSupported SOAP fault."""
27+
try:
28+
# Handle ONVIFOperationException
29+
if isinstance(exception, ONVIFOperationException):
30+
original = exception.original_exception
31+
else:
32+
original = exception
33+
34+
# Check if it's a Fault with subcodes
35+
if isinstance(original, Fault):
36+
subcodes = getattr(original, "subcodes", None)
37+
if subcodes:
38+
for subcode in subcodes:
39+
if hasattr(subcode, 'localname'):
40+
if subcode.localname == 'ActionNotSupported':
41+
return True
42+
elif 'ActionNotSupported' in str(subcode):
43+
return True
44+
except:
45+
pass
46+
47+
return False
48+
49+
@staticmethod
50+
def safe_call(func, default=None, ignore_unsupported=True, log_error=True):
51+
"""Safely call an ONVIF operation with graceful error handling."""
52+
try:
53+
return func()
54+
except ONVIFOperationException as e:
55+
# Check if it's ActionNotSupported error
56+
if ignore_unsupported and ONVIFErrorHandler.is_action_not_supported(e):
57+
#if log_error:
58+
#logging.warning(f"Operation not supported: {e.operation}")
59+
return default
60+
# Re-raise other errors
61+
raise
62+
except Exception as e:
63+
# Wrap unexpected exceptions
64+
#if log_error:
65+
#logging.error(f"Unexpected error in safe_call: {e}")
66+
raise
67+
68+
@staticmethod
69+
def ignore_unsupported(func):
70+
"""
71+
Decorator to ignore ActionNotSupported SOAP faults.
72+
Returns None for unsupported operations, raises other exceptions.
73+
"""
74+
def wrapper(*args, **kwargs):
75+
try:
76+
return func(*args, **kwargs)
77+
except ONVIFOperationException as e:
78+
if ONVIFErrorHandler.is_action_not_supported(e):
79+
#logging.warning(f"Operation not supported: {e.operation}")
80+
return None
81+
raise
82+
return wrapper
83+

0 commit comments

Comments
 (0)