ONVIFOperator
Low-level ONVIF service operator using Zeep SOAP client.
This class handles the actual SOAP communication with ONVIF devices. It manages WSDL loading, service binding, authentication, caching, and error handling.
Danger
ONVIFOperator is typically used internally by service classes
such as Device, Media, and PTZ, and is not intended to be
instantiated directly by end users.
Use ONVIFClient instead.
Attributes:
| Name | Type | Description |
|---|---|---|
wsdl_path |
str
|
Path to the WSDL file |
host |
str
|
Device hostname or IP address |
port |
int
|
Device port number |
username |
str | None
|
ONVIF username |
password |
str | None
|
ONVIF password |
http_digest |
bool
|
Whether to use HTTP Digest or WS-UsernameToken for auth |
timeout |
int
|
Request timeout in seconds |
apply_patch |
bool
|
Whether to apply |
address |
str
|
Service endpoint URL (XAddr) |
client |
Client
|
Zeep SOAP client instance |
service |
ServiceProxy
|
Zeep service proxy for making SOAP calls |
service_name |
str
|
Name of the ONVIF service (e.g., "Device", "Media") |
Version History
- Available since
>=v0.0.1(first release).
Source code in onvif\operator.py
class ONVIFOperator:
"""Low-level ONVIF service operator using Zeep SOAP client.
This class handles the actual SOAP communication with ONVIF devices. It manages
WSDL loading, service binding, authentication, caching, and error handling.
!!! danger
``ONVIFOperator`` is typically used internally by service classes
such as Device, Media, and PTZ, and is not intended to be
instantiated directly by end users.
Use [`ONVIFClient`](onvif_client.md) instead.
Attributes:
wsdl_path (str): Path to the WSDL file
host (str): Device hostname or IP address
port (int): Device port number
username (str | None): ONVIF username
password (str | None): ONVIF password
http_digest (bool): Whether to use **HTTP Digest** or **WS-UsernameToken** for auth
timeout (int): Request timeout in seconds
apply_patch (bool): Whether to apply ``xsd:any`` flattening patch
address (str): Service endpoint URL (XAddr)
client (Client): Zeep SOAP client instance
service (ServiceProxy): Zeep service proxy for making SOAP calls
service_name (str): Name of the ONVIF service (e.g., "Device", "Media")
!!! tip "Version History"
- Available since [`>=v0.0.1`](/onvif-python/releases/#v0.0.1) (first release).
"""
def __init__(
self,
wsdl_path: str,
host: str,
port: int,
username: str | None = None,
password: str | None = None,
http_digest: bool = False, # True = use HTTP Digest / False = use WS-UsernameToken
timeout: int = 10,
binding: str | None = None,
service_path: str | None = None,
xaddr: str | None = None,
cache: CacheMode = CacheMode.DB, # db | mem | none
cache_path: str | None = None,
use_https: bool = False,
verify_ssl: bool = False,
apply_patch: bool = True,
plugins: list | None = None,
):
logger.debug(
"Creating ONVIFOperator for %s:%d with WSDL: %s", host, port, wsdl_path
)
self.wsdl_path: str = wsdl_path
self.host: str = host
self.port: int = port
self.username: str | None = username
self.password: str | None = password
self.http_digest: bool = http_digest
self.timeout: int = timeout
self.apply_patch: bool = apply_patch
if xaddr:
self.address: str = xaddr
else:
protocol = "https" if use_https else "http"
path = service_path or "device_service" # default fallback
self.address = f"{protocol}://{self.host}:{self.port}/onvif/{path}"
logger.debug("Service endpoint: %s", self.address)
# Session reuse with retry strategy
session: Session = self._create_session(verify_ssl=verify_ssl)
transport_kwargs = {"session": session, "operation_timeout": self.timeout}
if cache == CacheMode.MEM:
logger.debug("Using in-memory WSDL cache")
transport_kwargs["cache"] = InMemoryCache()
elif cache == CacheMode.DB:
if cache_path is None:
user_cache_dir = os.path.expanduser("~/.onvif-python")
os.makedirs(user_cache_dir, exist_ok=True)
cache_path = os.path.join(user_cache_dir, "onvif_zeep_cache.sqlite")
logger.debug("Using SQLite cache: %s", cache_path)
transport_kwargs["cache"] = SqliteCache(path=cache_path)
elif cache != CacheMode.NONE:
raise ValueError(f"Unknown cache option: {cache}")
transport = Transport(**transport_kwargs)
# zeep settings
settings = Settings(strict=False, xml_huge_tree=True)
wsse: UsernameToken | None = self._create_wsse()
logger.debug("Using cache mode: %s", cache.value)
self.client: Client = Client(
wsdl=self.wsdl_path,
transport=transport,
settings=settings,
wsse=wsse,
plugins=plugins,
)
if not binding:
raise ValueError("Bindings must be set according to the WSDL service")
self.service: ServiceProxy = self.client.create_service(
binding_name=binding, address=self.address
)
self.service_name: str = binding.split("}")[-1].replace(
"Binding", ""
) # Store cleaned service name for logging context
logger.info("ONVIFOperator initialized %s at %s", binding, self.address)
def _create_session(
self,
verify_ssl: bool,
) -> Session:
"""Create and configure the HTTP session.
Args:
verify_ssl: Whether SSL certificates should be verified.
Returns:
Configured requests session.
"""
session = requests.Session()
session.verify = verify_ssl
if not verify_ssl:
# Format SSL warnings to be more concise when verify_ssl is False
logger.debug("SSL verification disabled")
warnings.simplefilter(
"once",
urllib3.exceptions.InsecureRequestWarning,
)
if self.http_digest and self.username and self.password:
logger.debug("Configuring HTTP Digest authentication")
session.auth = HTTPDigestAuth(
username=self.username,
password=self.password,
)
return session
def _create_wsse(
self,
) -> UsernameToken | None:
"""Create WS-Security authentication configuration.
Returns:
Zeep WS-Security UsernameToken or None.
"""
if self.http_digest:
return None
if not self.username or not self.password:
return None
logger.debug("Configuring WS-Security UsernameToken authentication")
return UsernameToken(
username=self.username,
password=self.password,
use_digest=True,
)
def call(self, method: str, *args, **kwargs) -> Any:
"""Call an ONVIF service operation.
This method invokes a SOAP operation on the ONVIF device service and handles
errors gracefully. It automatically flattens `xsd:any` fields in the response
when `apply_patch` is enabled.
Args:
method (str): Name of the ONVIF operation to call (e.g., "GetDeviceInformation")
*args (any): Positional arguments to pass to the operation
**kwargs (any): Keyword arguments to pass to the operation
Returns:
The operation result with `xsd:any` fields flattened if `apply_patch=True`
Raises:
ONVIFOperationException: If the operation fails (wraps original exception)
"""
logger.debug("Calling ONVIF method: %s.%s", self.service_name, method)
try:
func = getattr(self.service, method)
except AttributeError as e:
raise ONVIFOperationException(operation=method, original_exception=e) from e
try:
result = func(*args, **kwargs)
logger.debug("ONVIF call %s.%s succeeded", self.service_name, method)
# Post-process to flatten xsd:any fields if enabled (> v0.0.4 patch)
if self.apply_patch:
result = ZeepPatcher.flatten_xsd_any_fields(result)
return result
except Fault as e:
raise ONVIFOperationException(operation=method, original_exception=e) from e
except Exception as e:
raise ONVIFOperationException(operation=method, original_exception=e) from e
def create_type(self, type_name: str) -> Any:
"""Create a type instance from WSDL schema for the given type name.
Recursively initializes nested complex types so that fields like TimeZone, DateTime,
Date, and Time are properly instantiated as objects rather than None.
Args:
type_name (str): Name of the type to create (e.g., 'SetHostname', 'SetIPAddressFilter')
Returns:
Type instance that can be populated with data
Raises:
AttributeError: If type not found in WSDL schema
"""
logger.debug("Creating type instance for: %s", type_name)
# Method 1: Try to get element from WSDL (works for operation parameters)
# Common namespace prefixes for ONVIF services
namespaces_to_try = [
"ns0", # Default namespace
"tt", # Common types (User, NetworkInterface, etc.)
]
# Try to get element with namespace prefix
for ns in namespaces_to_try:
try:
element = self.client.get_element(f"{ns}:{type_name}")
instance = element()
logger.debug(
"Successfully created type %s using namespace %s", type_name, ns
)
return self._initialize_nested_types(instance)
except (AttributeError, TypeError, ValueError) as e:
logger.debug(
"Failed to create type %s with namespace %s: %s", type_name, ns, e
)
continue
# Method 2: Try without namespace prefix
try:
element = self.client.get_element(type_name)
instance = element()
logger.debug("Successfully created type %s without namespace", type_name)
return self._initialize_nested_types(instance)
except (AttributeError, TypeError, ValueError) as e:
logger.debug(
"Failed to create element %s without namespace: %s", type_name, e
)
# Method 3: Try to get type from schema (for complex types)
try:
for ns in namespaces_to_try:
try:
type_obj = self.client.get_type(f"{ns}:{type_name}")
instance = type_obj()
logger.debug(
"Successfully created complex type %s using namespace %s",
type_name,
ns,
)
return self._initialize_nested_types(instance)
except (AttributeError, TypeError, ValueError) as e:
logger.debug(
"Failed to create complex type %s with namespace %s: %s",
type_name,
ns,
e,
)
continue
# Try without namespace
type_obj = self.client.get_type(type_name)
instance = type_obj()
logger.debug(
"Successfully created complex type %s without namespace", type_name
)
return self._initialize_nested_types(instance)
except (AttributeError, TypeError, ValueError) as e:
logger.debug(
"Failed to create complex type %s without namespace: %s", type_name, e
)
# If all methods fail, log the error and raise
logger.error("Type '%s' not found in WSDL schema using any method", type_name)
raise AttributeError(f"Type '{type_name}' not found in WSDL schema.")
def _initialize_nested_types(self, instance):
"""Recursively initialize nested complex types in a Zeep object.
This ensures that fields like TimeZone, DateTime, Date, and Time are
properly instantiated as objects rather than None values.
Args:
instance: A Zeep object instance to initialize
Returns:
The instance with all nested complex types initialized
"""
# pylint: disable=too-many-nested-blocks
try:
# Get the XSD type from the instance's class
if hasattr(instance.__class__, "_xsd_type"):
xsd_type = (
instance.__class__._xsd_type # pylint: disable=protected-access
)
# Iterate through elements defined in the XSD type
if hasattr(xsd_type, "elements"):
for element_name, element_obj in xsd_type.elements:
current_value = getattr(instance, element_name, None)
# Only initialize if the value is None and the element has a type
if current_value is None and hasattr(element_obj, "type"):
element_type = element_obj.type
# Check if this is a complex type (has elements)
if hasattr(element_type, "elements"):
try:
# Complex type - instantiate it and recursively initialize
nested_instance = element_type()
nested_instance = self._initialize_nested_types(
nested_instance
)
setattr(instance, element_name, nested_instance)
logger.debug(
"Initialized nested type for element: %s",
element_name,
)
except (AttributeError, TypeError, ValueError) as e:
# Log specific nested type initialization failures
logger.debug(
"Failed to initialize nested type for %s: %s",
element_name,
e,
)
# Continue with other elements instead of failing completely
continue
except (AttributeError, TypeError, ValueError) as e:
# Log the error but don't fail - the top-level object is still usable
logger.debug("Error during nested type initialization: %s", e)
# The important thing is that the top-level object is created
return instance
call(method: str, *args, **kwargs) -> Any
Call an ONVIF service operation.
This method invokes a SOAP operation on the ONVIF device service and handles
errors gracefully. It automatically flattens xsd:any fields in the response
when apply_patch is enabled.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
method
|
str
|
Name of the ONVIF operation to call (e.g., "GetDeviceInformation") |
required |
*args
|
any
|
Positional arguments to pass to the operation |
()
|
**kwargs
|
any
|
Keyword arguments to pass to the operation |
{}
|
Returns:
| Type | Description |
|---|---|
Any
|
The operation result with |
Raises:
| Type | Description |
|---|---|
ONVIFOperationException
|
If the operation fails (wraps original exception) |
Source code in onvif\operator.py
def call(self, method: str, *args, **kwargs) -> Any:
"""Call an ONVIF service operation.
This method invokes a SOAP operation on the ONVIF device service and handles
errors gracefully. It automatically flattens `xsd:any` fields in the response
when `apply_patch` is enabled.
Args:
method (str): Name of the ONVIF operation to call (e.g., "GetDeviceInformation")
*args (any): Positional arguments to pass to the operation
**kwargs (any): Keyword arguments to pass to the operation
Returns:
The operation result with `xsd:any` fields flattened if `apply_patch=True`
Raises:
ONVIFOperationException: If the operation fails (wraps original exception)
"""
logger.debug("Calling ONVIF method: %s.%s", self.service_name, method)
try:
func = getattr(self.service, method)
except AttributeError as e:
raise ONVIFOperationException(operation=method, original_exception=e) from e
try:
result = func(*args, **kwargs)
logger.debug("ONVIF call %s.%s succeeded", self.service_name, method)
# Post-process to flatten xsd:any fields if enabled (> v0.0.4 patch)
if self.apply_patch:
result = ZeepPatcher.flatten_xsd_any_fields(result)
return result
except Fault as e:
raise ONVIFOperationException(operation=method, original_exception=e) from e
except Exception as e:
raise ONVIFOperationException(operation=method, original_exception=e) from e
create_type(type_name: str) -> Any
Create a type instance from WSDL schema for the given type name.
Recursively initializes nested complex types so that fields like TimeZone, DateTime, Date, and Time are properly instantiated as objects rather than None.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
type_name
|
str
|
Name of the type to create (e.g., 'SetHostname', 'SetIPAddressFilter') |
required |
Returns:
| Type | Description |
|---|---|
Any
|
Type instance that can be populated with data |
Raises:
| Type | Description |
|---|---|
AttributeError
|
If type not found in WSDL schema |
Source code in onvif\operator.py
def create_type(self, type_name: str) -> Any:
"""Create a type instance from WSDL schema for the given type name.
Recursively initializes nested complex types so that fields like TimeZone, DateTime,
Date, and Time are properly instantiated as objects rather than None.
Args:
type_name (str): Name of the type to create (e.g., 'SetHostname', 'SetIPAddressFilter')
Returns:
Type instance that can be populated with data
Raises:
AttributeError: If type not found in WSDL schema
"""
logger.debug("Creating type instance for: %s", type_name)
# Method 1: Try to get element from WSDL (works for operation parameters)
# Common namespace prefixes for ONVIF services
namespaces_to_try = [
"ns0", # Default namespace
"tt", # Common types (User, NetworkInterface, etc.)
]
# Try to get element with namespace prefix
for ns in namespaces_to_try:
try:
element = self.client.get_element(f"{ns}:{type_name}")
instance = element()
logger.debug(
"Successfully created type %s using namespace %s", type_name, ns
)
return self._initialize_nested_types(instance)
except (AttributeError, TypeError, ValueError) as e:
logger.debug(
"Failed to create type %s with namespace %s: %s", type_name, ns, e
)
continue
# Method 2: Try without namespace prefix
try:
element = self.client.get_element(type_name)
instance = element()
logger.debug("Successfully created type %s without namespace", type_name)
return self._initialize_nested_types(instance)
except (AttributeError, TypeError, ValueError) as e:
logger.debug(
"Failed to create element %s without namespace: %s", type_name, e
)
# Method 3: Try to get type from schema (for complex types)
try:
for ns in namespaces_to_try:
try:
type_obj = self.client.get_type(f"{ns}:{type_name}")
instance = type_obj()
logger.debug(
"Successfully created complex type %s using namespace %s",
type_name,
ns,
)
return self._initialize_nested_types(instance)
except (AttributeError, TypeError, ValueError) as e:
logger.debug(
"Failed to create complex type %s with namespace %s: %s",
type_name,
ns,
e,
)
continue
# Try without namespace
type_obj = self.client.get_type(type_name)
instance = type_obj()
logger.debug(
"Successfully created complex type %s without namespace", type_name
)
return self._initialize_nested_types(instance)
except (AttributeError, TypeError, ValueError) as e:
logger.debug(
"Failed to create complex type %s without namespace: %s", type_name, e
)
# If all methods fail, log the error and raise
logger.error("Type '%s' not found in WSDL schema using any method", type_name)
raise AttributeError(f"Type '{type_name}' not found in WSDL schema.")