Skip to content

Commit b450f41

Browse files
committed
code(examples): Add comprehensive logging demonstration script for ONVIF operations
1 parent a187e78 commit b450f41

File tree

1 file changed

+353
-0
lines changed

1 file changed

+353
-0
lines changed

examples/logger.py

Lines changed: 353 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,353 @@
1+
"""
2+
Path: examples/logger.py
3+
Author: @kaburagisec
4+
Created: October 29, 2025
5+
Tested devices: EZVIZ H8C (https://www.ezviz.com/inter/product/h8c/43162)
6+
7+
Applied for (>= v0.2.0 patch)
8+
9+
This script demonstrates the comprehensive logging functionality implemented
10+
across the ONVIF Python library. It shows how to configure different logging
11+
levels and capture detailed information about ONVIF operations.
12+
13+
Usage:
14+
python logger.py [device_ip] [username] [password] [options]
15+
16+
Example:
17+
python logger.py 192.168.1.17 admin admin123 --port 8000
18+
python logger.py --discover-only # Just run discovery
19+
python logger.py --level DEBUG # Will show all logs
20+
python logger.py --level WARNING # Only warnings and errors
21+
python logger.py --level ERROR # Only errors messages
22+
python logger.py --level INFO # Only info, errors, warnings
23+
"""
24+
25+
import sys
26+
import logging
27+
import argparse
28+
from datetime import datetime
29+
import os
30+
31+
from onvif import ONVIFClient, ONVIFDiscovery, ONVIFWSDL
32+
33+
34+
def setup_logging(level=logging.INFO):
35+
"""Configure logging with detailed formatting for ONVIF operations.
36+
37+
Args:
38+
level: Logging level (DEBUG, INFO, WARNING, ERROR)
39+
"""
40+
# Create formatter
41+
formatter = logging.Formatter(
42+
"%(asctime)s - %(name)s - %(levelname)s - %(message)s",
43+
datefmt="%Y-%m-%d %H:%M:%S",
44+
)
45+
46+
# Console handler
47+
console_handler = logging.StreamHandler(sys.stdout)
48+
console_handler.setLevel(level)
49+
console_handler.setFormatter(formatter)
50+
51+
# File handler for detailed logging
52+
log_filename = (
53+
f"onvif_logging_example_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log"
54+
)
55+
file_handler = logging.FileHandler(log_filename)
56+
file_handler.setLevel(level)
57+
file_handler.setFormatter(formatter)
58+
59+
# Configure ONVIF logger (this will handle all onvif.* loggers)
60+
onvif_logger = logging.getLogger("onvif")
61+
onvif_logger.setLevel(level)
62+
onvif_logger.addHandler(console_handler)
63+
onvif_logger.addHandler(file_handler)
64+
65+
# Suppress verbose logging from external libraries
66+
for logger_name in ["zeep", "urllib3", "requests"]:
67+
logging.getLogger(logger_name).setLevel(logging.WARNING)
68+
69+
print(
70+
f"Logging configured - Level: {logging.getLevelName(level)}, File: {log_filename}"
71+
)
72+
return log_filename
73+
74+
75+
def demonstrate_discovery_logging():
76+
"""Demonstrate ONVIF device discovery with detailed logging."""
77+
print("\n" + "=" * 60)
78+
print("ONVIF DEVICE DISCOVERY LOGGING DEMONSTRATION")
79+
print("=" * 60)
80+
81+
# Create discovery instance with logging
82+
discovery = ONVIFDiscovery(timeout=5)
83+
84+
print("\n1. Starting device discovery...")
85+
devices = discovery.discover(prefer_https=True, search=None)
86+
87+
print(f"\n2. Discovery Results:")
88+
if devices:
89+
for i, device in enumerate(devices, 1):
90+
print(f" Device {i}:")
91+
print(f" Host: {device['host']}")
92+
print(f" Port: {device['port']}")
93+
print(f" Endpoint: {device['host']}:{device['port']}")
94+
print(f" HTTPS: {device['use_https']}")
95+
print(f" Types: {', '.join(device.get('types', []))}")
96+
print(
97+
f" Scopes: {', '.join(device.get('scopes', [])[:2])}..."
98+
) # First 2 scopes
99+
else:
100+
print(" No devices found")
101+
102+
return devices
103+
104+
105+
def demonstrate_wsdl_logging():
106+
"""Demonstrate WSDL resolution and management logging."""
107+
print("\n" + "=" * 60)
108+
print("WSDL RESOLUTION LOGGING DEMONSTRATION")
109+
print("=" * 60)
110+
111+
print("\n1. Getting WSDL definitions...")
112+
113+
# Test various WSDL services
114+
services_to_test = [
115+
("devicemgmt", "ver10"),
116+
("media", "ver20"), # This will raise ERROR log (wrong version for testing)
117+
("ptz", "ver20"),
118+
("imaging", "ver20"),
119+
("events", "ver10"),
120+
]
121+
122+
for service, version in services_to_test:
123+
try:
124+
definition = ONVIFWSDL.get_definition(service, version)
125+
print(f" ✓ {service} ({version}): {os.path.basename(definition['path'])}")
126+
except Exception as e:
127+
print(f" ✗ {service} ({version}): {e}")
128+
129+
# Show custom WSDL directory functionality
130+
print(
131+
f"\n2. Current WSDL directory: {ONVIFWSDL.get_custom_wsdl_dir() or 'Built-in'}"
132+
)
133+
134+
135+
def demonstrate_client_logging(host, port, username, password):
136+
"""Demonstrate ONVIF client operations with comprehensive logging."""
137+
print("\n" + "=" * 60)
138+
print("ONVIF CLIENT OPERATIONS LOGGING DEMONSTRATION")
139+
print("=" * 60)
140+
141+
print(f"\n1. Connecting to ONVIF device: {host}:{port}")
142+
143+
# Create client with XML capture for detailed SOAP logging
144+
client = ONVIFClient(
145+
host=host,
146+
port=port,
147+
username=username,
148+
password=password,
149+
timeout=10,
150+
capture_xml=True, # Enable XML capture
151+
use_https=False,
152+
verify_ssl=True,
153+
)
154+
155+
print("\n2. Accessing Device Management service...")
156+
try:
157+
device = client.devicemgmt()
158+
print(" ✓ Device Management service initialized")
159+
160+
# Test basic device operations
161+
print("\n3. Getting device information...")
162+
device_info = device.GetDeviceInformation()
163+
print(f" ✓ Device: {device_info.Manufacturer} {device_info.Model}")
164+
print(f" ✓ Firmware: {device_info.FirmwareVersion}")
165+
print(f" ✓ Serial: {device_info.SerialNumber}")
166+
167+
except Exception as e:
168+
print(f" ✗ Device Management failed: {e}")
169+
return None
170+
171+
print("\n4. Testing service capabilities...")
172+
173+
# Test various services with error handling
174+
services_to_test = [
175+
("media", "GetProfiles"),
176+
("ptz", "GetConfigurations"),
177+
("imaging", "GetImagingSettings"),
178+
("events", "GetEventProperties"),
179+
]
180+
181+
for service_name, operation in services_to_test:
182+
try:
183+
print(f" Testing {service_name}.{operation}...")
184+
service = getattr(client, service_name)()
185+
method = getattr(service, operation)
186+
result = method()
187+
print(f" ✓ {service_name}.{operation} succeeded")
188+
except Exception as e:
189+
print(f" ✗ {service_name}.{operation} failed: {type(e).__name__}")
190+
191+
print("\n5. Testing type creation...")
192+
try:
193+
# Test type creation with logging
194+
hostname_type = device.type("SetHostname")
195+
print(" ✓ SetHostname type created successfully")
196+
197+
user_type = device.type("CreateUsers")
198+
print(" ✓ CreateUsers type created successfully")
199+
200+
except Exception as e:
201+
print(f" ✗ Type creation failed: {e}")
202+
203+
# Show XML capture results if available
204+
if hasattr(client, "xml_plugin") and client.xml_plugin:
205+
print(f"\n6. XML Capture Summary:")
206+
history = client.xml_plugin.get_history()
207+
print(f" Total SOAP transactions: {len(history)}")
208+
209+
if history:
210+
last_request = client.xml_plugin.get_last_request()
211+
last_response = client.xml_plugin.get_last_response()
212+
print(
213+
f" Last request size: {len(last_request) if last_request else 0} chars"
214+
)
215+
print(
216+
f" Last response size: {len(last_response) if last_response else 0} chars"
217+
)
218+
219+
return client
220+
221+
222+
def demonstrate_error_handling_logging():
223+
"""Demonstrate error handling and exception logging."""
224+
print("\n" + "=" * 60)
225+
print("ERROR HANDLING LOGGING DEMONSTRATION")
226+
print("=" * 60)
227+
228+
print("\n1. Testing connection to non-existent device...")
229+
try:
230+
# This should fail and demonstrate error logging
231+
fake_client = ONVIFClient(
232+
host="192.168.999.999", port=80, username="fake", password="fake", timeout=3
233+
)
234+
device = fake_client.devicemgmt()
235+
device.GetDeviceInformation()
236+
except Exception as e:
237+
print(f" ✓ Expected error logged: {type(e).__name__}")
238+
239+
print("\n2. Testing invalid service operations...")
240+
# Additional error scenarios could be added here
241+
print(" ✓ Error handling demonstration complete")
242+
243+
244+
def main():
245+
"""Main demonstration function with command line argument handling."""
246+
parser = argparse.ArgumentParser(
247+
description="ONVIF Library Logging Demonstration",
248+
formatter_class=argparse.RawDescriptionHelpFormatter,
249+
epilog=__doc__,
250+
)
251+
252+
parser.add_argument("host", nargs="?", default=None, help="ONVIF device IP address")
253+
parser.add_argument(
254+
"username", nargs="?", default="admin", help="ONVIF username (default: admin)"
255+
)
256+
parser.add_argument(
257+
"password", nargs="?", default="admin", help="ONVIF password (default: admin)"
258+
)
259+
parser.add_argument(
260+
"--port", type=int, default=80, help="ONVIF device port (default: 80)"
261+
)
262+
parser.add_argument(
263+
"--level",
264+
choices=["DEBUG", "INFO", "WARNING", "ERROR"],
265+
default="INFO",
266+
help="Logging level (default: INFO)",
267+
)
268+
parser.add_argument(
269+
"--discover-only", action="store_true", help="Only run device discovery"
270+
)
271+
272+
args = parser.parse_args()
273+
274+
# Setup logging
275+
log_level = getattr(logging, args.level.upper())
276+
log_file = setup_logging(log_level)
277+
278+
print("ONVIF Python Library - Comprehensive Logging Example")
279+
print(f"Timestamp: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
280+
print(f"Log Level: {log_level}")
281+
282+
try:
283+
# Always demonstrate discovery
284+
devices = demonstrate_discovery_logging()
285+
286+
# Always demonstrate WSDL logging
287+
demonstrate_wsdl_logging()
288+
289+
if not args.discover_only:
290+
# Determine device to connect to
291+
if args.host:
292+
host = args.host
293+
port = args.port
294+
elif devices:
295+
# Use first discovered device
296+
device = devices[0]
297+
host = device["host"]
298+
port = device["port"]
299+
print(f"\nUsing discovered device: {host}:{port}")
300+
else:
301+
print(
302+
"\nNo device specified and none discovered. Use --discover-only or provide device details."
303+
)
304+
return
305+
306+
# Demonstrate client operations
307+
client = demonstrate_client_logging(
308+
host, port, args.username, args.password
309+
)
310+
311+
# Demonstrate error handling
312+
demonstrate_error_handling_logging()
313+
314+
except KeyboardInterrupt:
315+
print("\n\nDemo interrupted by user")
316+
except Exception as e:
317+
print(f"\nUnexpected error: {e}")
318+
logging.exception("Unexpected error in main demonstration")
319+
320+
finally:
321+
print("\n" + "=" * 60)
322+
print("LOGGING DEMONSTRATION COMPLETE")
323+
print("=" * 60)
324+
print(f"\nDetailed logs saved to: {log_file}")
325+
print("\nLog file contains:")
326+
print("- Complete ONVIF operation traces")
327+
print("- Debug information (ONVIF-specific)")
328+
print("- Error details and stack traces")
329+
print("- Network communication logs (ONVIF-focused)")
330+
print("- SOAP XML requests/responses (if XML capture enabled)")
331+
print("- Filtered output (excludes verbose Zeep/HTTP library logs)")
332+
333+
# Show brief log file summary
334+
try:
335+
with open(log_file, "r", encoding="utf-8") as f:
336+
lines = f.readlines()
337+
338+
debug_count = sum(1 for line in lines if " - DEBUG - " in line)
339+
info_count = sum(1 for line in lines if " - INFO - " in line)
340+
warning_count = sum(1 for line in lines if " - WARNING - " in line)
341+
error_count = sum(1 for line in lines if " - ERROR - " in line)
342+
343+
print(f"\nLog Summary: {len(lines)} total lines")
344+
print(
345+
f" DEBUG: {debug_count}, INFO: {info_count}, WARNING: {warning_count}, ERROR: {error_count}"
346+
)
347+
348+
except Exception:
349+
pass
350+
351+
352+
if __name__ == "__main__":
353+
main()

0 commit comments

Comments
 (0)