ONVIFClient
onvif.client.ONVIFClient
ONVIFClient(
host: str,
port: int,
username: str | None = None,
password: str | None = None,
http_digest: bool = False,
timeout: int = 10,
cache: CacheMode = DB,
use_https: bool = False,
verify_ssl: bool = False,
apply_patch: bool = True,
capture_xml: bool = False,
wsdl_dir: str | None = None,
plugins: list[Plugin] | None = None,
)
High-level ONVIF client for interacting with ONVIF-compliant devices.
This is the main class for interacting with ONVIF devices. It provides access to all ONVIF services including Device Management, Media, PTZ, Events, Analytics, and more.
The client automatically discovers available services on the device using GetServices or GetCapabilities, and provides lazy initialization for service endpoints.
Attributes:
| Name | Type | Description |
|---|---|---|
services | list | None | List of available services from |
capabilities | CompoundValue | None | Device capabilities from |
xml_plugin | XMLCapturePlugin | None | XML capture plugin for debugging (if |
wsdl_dir | str | None | Custom WSDL directory path (if provided) |
plugins | list[Plugin] | None | List of user-provided Zeep plugins (if any) |
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
host | str | Device hostname or IP address | required |
port | int | Device port number | required |
username | str | None | ONVIF username | None |
password | str | None | ONVIF password | None |
http_digest | bool | Whether to use HTTP Digest or WS-UsernameToken for auth | False |
timeout | int | Request timeout in seconds | 10 |
cache | CacheMode | WSDL caching strategy | DB |
use_https | bool | Use HTTPS instead of HTTP for secure communication | False |
verify_ssl | bool | Whether SSL certificates should be verified | False |
apply_patch | bool | Whether to apply | True |
capture_xml | bool | Whether to use XML capture plugin for debugging SOAP requests/responses | False |
wsdl_dir | str | None | Custom WSDL directory path for using external WSDL files instead of built-in ones | None |
plugins | list[Plugin] | None | List of user-provided Zeep plugins ( | None |
Version History
Source code in onvif/client.py
def __init__(
self,
host: str,
port: int,
username: str | None = None,
password: str | None = None,
http_digest: bool = False, # will use WS-UsernameToken by default
timeout: int = 10,
cache: CacheMode = CacheMode.DB,
use_https: bool = False,
verify_ssl: bool = False,
apply_patch: bool = True,
capture_xml: bool = False,
wsdl_dir: str | None = None,
plugins: list[Plugin] | None = None,
):
"""Initialize the ONVIF client.
Args:
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
cache (CacheMode): WSDL caching strategy
use_https (bool): Use HTTPS instead of HTTP for secure communication
verify_ssl (bool): Whether SSL certificates should be verified
apply_patch (bool): Whether to apply ``xsd:any`` flattening patch
capture_xml (bool): Whether to use XML capture plugin for debugging SOAP requests/responses
wsdl_dir (str | None): Custom WSDL directory path for using external WSDL files instead of built-in ones
plugins (list[Plugin] | None): List of user-provided Zeep plugins (`zeep.Plugin`)
!!! tip "Version History"
- Added in [`v0.0.4`](/onvif-python/releases/#v0.0.4): `apply_patch`
- Added in [`v0.0.6`](/onvif-python/releases/#v0.0.6): `capture_xml`
- Added in [`v0.1.0`](/onvif-python/releases/#v0.1.0): `wsdl_dir`
- Added in [`v0.2.2`](/onvif-python/releases/#v0.2.2): `plugins`
- Added in [`v0.3.0`](/onvif-python/releases/#v0.3.0): `http_digest`
- Changed in [`v0.3.0`](/onvif-python/releases/#v0.3.0): `username` → `str | None`,
`password` → `str | None`
"""
logger.info("Initializing ONVIF client for %s:%d", host, port)
logger.debug(
"Connection settings: HTTPS=%s, SSL_verify=%s, cache=%s, timeout=%ds",
use_https,
verify_ssl,
cache.value,
timeout,
)
# Apply or remove zeep patch based on user preference
self._configure_patches(apply_patch)
# Initialize XML capture plugin if requested
# and merge user plugins with xml_plugin
all_plugins = self._configure_plugins(
capture_xml,
plugins,
)
self.plugins = plugins
# Store custom WSDL directory if provided
self.wsdl_dir: str | None = wsdl_dir
if wsdl_dir:
logger.debug("Using custom WSDL directory: %s", wsdl_dir)
ONVIFWSDL.set_custom_wsdl_dir(wsdl_dir)
# Pass to ONVIFOperator
self.common_args = {
"host": host,
"port": port,
"username": username,
"password": password,
"http_digest": http_digest,
"timeout": timeout,
"cache": cache,
"use_https": use_https,
"verify_ssl": verify_ssl,
"apply_patch": apply_patch,
"plugins": all_plugins if all_plugins else None,
}
# Device Management (Core) service is always available
self._devicemgmt: Device | None = None
self._devicemgmt = self.devicemgmt()
# Try to retrieve device services and create namespace -> XAddr mapping
self.services = None
self._service_map = {}
# Temporary variable to hold capabilities
self.capabilities = None
try:
# Try GetServices first (preferred method)
logger.debug("Attempting GetServices call for service discovery")
self.services = self._devicemgmt.GetServices(IncludeCapability=False)
logger.info("Found %d services via GetServices", len(self.services))
for onvif_service in self.services:
namespace = getattr(onvif_service, "Namespace", None)
xaddr = getattr(onvif_service, "XAddr", None)
if namespace and xaddr:
self._service_map[namespace] = xaddr
logger.debug("Mapped service: %s -> %s", namespace, xaddr)
except (ValueError, AttributeError, ONVIFOperationException) as e:
logger.warning("GetServices failed: %s", e)
# Fallback to GetCapabilities if GetServices is not supported on device
try:
logger.debug("Falling back to GetCapabilities")
self.capabilities = self._devicemgmt.GetCapabilities(Category="All")
logger.info("Successfully retrieved device capabilities")
except ONVIFOperationException as e2:
# If both fail, we'll use default URLs
logger.error("Both GetServices and GetCapabilities failed: %s", e2)
logger.warning("Using default URLs for services")
# Lazy init for other services
self._events: Events | None = None
# Dictionary for multiple PullPoint instances
self._pullpoints: dict[str, PullPoint] = {}
self._notification: Notification | None = None
# Dictionary for multiple Subscription instances
self._subscriptions: dict[str, Subscription] = {}
# Dictionary for multiple PausableSubscription instances
self._pausable_subscriptions: dict[str, PausableSubscription] = {}
self._imaging: Imaging | None = None
self._media: Media | None = None
self._media2: Media2 | None = None
self._ptz: PTZ | None = None
self._deviceio: DeviceIO | None = None
self._display: Display | None = None
self._analytics: Analytics | None = None
self._ruleengine: RuleEngine | None = None
self._analyticsdevice: AnalyticsDevice | None = None
self._accesscontrol: AccessControl | None = None
self._doorcontrol: DoorControl | None = None
self._accessrules: AccessRules | None = None
self._actionengine: ActionEngine | None = None
self._appmanagement: AppManagement | None = None
self._authenticationbehavior: AuthenticationBehavior | None = None
self._credential: Credential | None = None
self._recording: Recording | None = None
self._replay: Replay | None = None
self._provisioning: Provisioning | None = None
self._receiver: Receiver | None = None
self._schedule: Schedule | None = None
self._search: Search | None = None
self._thermal: Thermal | None = None
self._uplink: Uplink | None = None
self._security: AdvancedSecurity | None = None
self._jwt: JWT | None = None
self._keystore: Keystore | None = None
self._tlsserver: TLSServer | None = None
self._dot1x: Dot1X | None = None
self._authorizationserver: AuthorizationServer | None = None
self._mediasigning: MediaSigning | None = None
devicemgmt
devicemgmt() -> Device
Access the Device Management service.
Source code in onvif/client.py
pullpoint
pullpoint(SubscriptionRef) -> PullPoint
Access the PullPoint service.
Source code in onvif/client.py
notification
notification() -> Notification
Access the Notification service.
Source code in onvif/client.py
@service
def notification(self) -> Notification:
"""
<!-- md:version 0.0.1 -->
<!-- md:entry_point -->
Access the Notification service."""
if self._notification is None:
logger.debug("Initializing Notification service")
self._notification = Notification(
xaddr=self._get_xaddr("notification", "Events"), **self.common_args
)
return self._notification
subscription
subscription(SubscriptionRef) -> Subscription
Access the Subscription service.
Source code in onvif/client.py
pausable_subscription
pausable_subscription(
SubscriptionRef,
) -> PausableSubscription
Access the PausableSubscription service.
Source code in onvif/client.py
@service
def pausable_subscription(self, SubscriptionRef) -> PausableSubscription:
"""
<!-- md:version 0.2.4 -->
<!-- md:entry_point -->
Access the PausableSubscription service.
"""
logger.debug("Initializing PausableSubscription service")
return self._get_subscription_service(
SubscriptionRef,
PausableSubscription,
self._pausable_subscriptions,
)
imaging
imaging() -> Imaging
Access the Imaging service.
Source code in onvif/client.py
@service
def imaging(self) -> Imaging:
"""
<!-- md:version 0.0.1 -->
<!-- md:entry_point -->
Access the Imaging service."""
if self._imaging is None:
logger.debug("Initializing Imaging service")
self._imaging = Imaging(
xaddr=self._get_xaddr("imaging", "Imaging"), **self.common_args
)
return self._imaging
deviceio
deviceio() -> DeviceIO
Access the DeviceIO service.
Source code in onvif/client.py
@service
def deviceio(self) -> DeviceIO:
"""
<!-- md:version 0.0.1 -->
<!-- md:entry_point -->
Access the DeviceIO service."""
if self._deviceio is None:
logger.debug("Initializing DeviceIO service")
self._deviceio = DeviceIO(
xaddr=self._get_xaddr("deviceio", "DeviceIO"), **self.common_args
)
return self._deviceio
display
display() -> Display
Access the Display service.
Source code in onvif/client.py
@service
def display(self) -> Display:
"""
<!-- md:version 0.0.1 -->
<!-- md:entry_point -->
Access the Display service."""
if self._display is None:
logger.debug("Initializing Display service")
self._display = Display(
xaddr=self._get_xaddr("display", "Display"), **self.common_args
)
return self._display
analytics
analytics() -> Analytics
Access the Analytics service.
Source code in onvif/client.py
@service
def analytics(self) -> Analytics:
"""
<!-- md:version 0.0.1 -->
<!-- md:entry_point -->
Access the Analytics service."""
if self._analytics is None:
logger.debug("Initializing Analytics service")
self._analytics = Analytics(
xaddr=self._get_xaddr("analytics", "Analytics"), **self.common_args
)
return self._analytics
ruleengine
ruleengine() -> RuleEngine
Access the RuleEngine service.
Source code in onvif/client.py
@service
def ruleengine(self) -> RuleEngine:
"""
<!-- md:version 0.0.1 -->
<!-- md:entry_point -->
Access the RuleEngine service."""
if self._ruleengine is None:
logger.debug("Initializing RuleEngine service")
self._ruleengine = RuleEngine(
xaddr=self._get_xaddr("ruleengine", "Analytics"), **self.common_args
)
return self._ruleengine
analyticsdevice
analyticsdevice() -> AnalyticsDevice
Access the AnalyticsDevice service.
Source code in onvif/client.py
@service
def analyticsdevice(self) -> AnalyticsDevice:
"""
<!-- md:version 0.0.1 -->
<!-- md:entry_point -->
Access the AnalyticsDevice service."""
if self._analyticsdevice is None:
logger.debug("Initializing AnalyticsDevice service")
self._analyticsdevice = AnalyticsDevice(
xaddr=self._get_xaddr("analyticsdevice", "AnalyticsDevice"),
**self.common_args,
)
return self._analyticsdevice
accesscontrol
accesscontrol() -> AccessControl
Access the AccessControl service.
Source code in onvif/client.py
@service
def accesscontrol(self) -> AccessControl:
"""
<!-- md:version 0.0.1 -->
<!-- md:entry_point -->
Access the AccessControl service."""
if self._accesscontrol is None:
logger.debug("Initializing AccessControl service")
self._accesscontrol = AccessControl(
xaddr=self._get_xaddr("accesscontrol", "AccessControl"),
**self.common_args,
)
return self._accesscontrol
doorcontrol
doorcontrol() -> DoorControl
Access the DoorControl service.
Source code in onvif/client.py
@service
def doorcontrol(self) -> DoorControl:
"""
<!-- md:version 0.0.1 -->
<!-- md:entry_point -->
Access the DoorControl service."""
if self._doorcontrol is None:
logger.debug("Initializing DoorControl service")
self._doorcontrol = DoorControl(
xaddr=self._get_xaddr("doorcontrol", "DoorControl"), **self.common_args
)
return self._doorcontrol
accessrules
accessrules() -> AccessRules
Access the AccessRules service.
Source code in onvif/client.py
@service
def accessrules(self) -> AccessRules:
"""
<!-- md:version 0.0.1 -->
<!-- md:entry_point -->
Access the AccessRules service."""
if self._accessrules is None:
logger.debug("Initializing AccessRules service")
self._accessrules = AccessRules(
xaddr=self._get_xaddr("accessrules", "AccessRules"), **self.common_args
)
return self._accessrules
actionengine
actionengine() -> ActionEngine
Access the ActionEngine service.
Source code in onvif/client.py
@service
def actionengine(self) -> ActionEngine:
"""
<!-- md:version 0.0.1 -->
<!-- md:entry_point -->
Access the ActionEngine service."""
if self._actionengine is None:
logger.debug("Initializing ActionEngine service")
self._actionengine = ActionEngine(
xaddr=self._get_xaddr("actionengine", "ActionEngine"),
**self.common_args,
)
return self._actionengine
appmanagement
appmanagement() -> AppManagement
Access the AppManagement service.
Source code in onvif/client.py
@service
def appmanagement(self) -> AppManagement:
"""
<!-- md:version 0.0.1 -->
<!-- md:entry_point -->
Access the AppManagement service."""
if self._appmanagement is None:
logger.debug("Initializing AppManagement service")
self._appmanagement = AppManagement(
xaddr=self._get_xaddr("appmgmt", "AppManagement"),
**self.common_args,
)
return self._appmanagement
authenticationbehavior
authenticationbehavior() -> AuthenticationBehavior
Access the AuthenticationBehavior service.
Source code in onvif/client.py
@service
def authenticationbehavior(self) -> AuthenticationBehavior:
"""
<!-- md:version 0.0.1 -->
<!-- md:entry_point -->
Access the AuthenticationBehavior service."""
if self._authenticationbehavior is None:
logger.debug("Initializing AuthenticationBehavior service")
self._authenticationbehavior = AuthenticationBehavior(
xaddr=self._get_xaddr(
"authenticationbehavior", "AuthenticationBehavior"
),
**self.common_args,
)
return self._authenticationbehavior
credential
credential() -> Credential
Access the Credential service.
Source code in onvif/client.py
@service
def credential(self) -> Credential:
"""
<!-- md:version 0.0.1 -->
<!-- md:entry_point -->
Access the Credential service."""
if self._credential is None:
logger.debug("Initializing Credential service")
self._credential = Credential(
xaddr=self._get_xaddr("credential", "Credential"),
**self.common_args,
)
return self._credential
recording
recording() -> Recording
Access the Recording service.
Source code in onvif/client.py
@service
def recording(self) -> Recording:
"""
<!-- md:version 0.0.1 -->
<!-- md:entry_point -->
Access the Recording service."""
if self._recording is None:
logger.debug("Initializing Recording service")
self._recording = Recording(
xaddr=self._get_xaddr("recording", "Recording"),
**self.common_args,
)
return self._recording
provisioning
provisioning() -> Provisioning
Access the Provisioning service.
Source code in onvif/client.py
@service
def provisioning(self) -> Provisioning:
"""
<!-- md:version 0.0.1 -->
<!-- md:entry_point -->
Access the Provisioning service."""
if self._provisioning is None:
logger.debug("Initializing Provisioning service")
self._provisioning = Provisioning(
xaddr=self._get_xaddr("provisioning", "Provisioning"),
**self.common_args,
)
return self._provisioning
receiver
receiver() -> Receiver
Access the Receiver service.
Source code in onvif/client.py
@service
def receiver(self) -> Receiver:
"""
<!-- md:version 0.0.1 -->
<!-- md:entry_point -->
Access the Receiver service."""
if self._receiver is None:
logger.debug("Initializing Receiver service")
self._receiver = Receiver(
xaddr=self._get_xaddr("receiver", "Receiver"),
**self.common_args,
)
return self._receiver
schedule
schedule() -> Schedule
Access the Schedule service.
Source code in onvif/client.py
@service
def schedule(self) -> Schedule:
"""
<!-- md:version 0.0.1 -->
<!-- md:entry_point -->
Access the Schedule service."""
if self._schedule is None:
logger.debug("Initializing Schedule service")
self._schedule = Schedule(
xaddr=self._get_xaddr("schedule", "Schedule"),
**self.common_args,
)
return self._schedule
thermal
thermal() -> Thermal
Access the Thermal service.
Source code in onvif/client.py
@service
def thermal(self) -> Thermal:
"""
<!-- md:version 0.0.1 -->
<!-- md:entry_point -->
Access the Thermal service."""
if self._thermal is None:
logger.debug("Initializing Thermal service")
self._thermal = Thermal(
xaddr=self._get_xaddr("thermal", "Thermal"),
**self.common_args,
)
return self._thermal
security
security() -> AdvancedSecurity
Access the AdvancedSecurity service.
Source code in onvif/client.py
@service
def security(self) -> AdvancedSecurity:
"""
<!-- md:version 0.0.1 -->
<!-- md:entry_point -->
Access the AdvancedSecurity service."""
if self._security is None:
logger.debug("Initializing Security service")
self._security = AdvancedSecurity(
xaddr=self._get_xaddr("advancedsecurity", "Security"),
**self.common_args,
)
return self._security
keystore
keystore() -> Keystore
Access the Keystore service.
Source code in onvif/client.py
@service
def keystore(self) -> Keystore:
"""
<!-- md:version 0.0.1 -->
<!-- md:entry_point -->
Access the Keystore service."""
if self._keystore is None:
logger.debug("Initializing Keystore service")
self._keystore = Keystore(
xaddr=self._get_xaddr("keystore", "Security"), **self.common_args
)
return self._keystore
tlsserver
tlsserver() -> TLSServer
Access the TLSServer service.
Source code in onvif/client.py
@service
def tlsserver(self) -> TLSServer:
"""
<!-- md:version 0.0.1 -->
<!-- md:entry_point -->
Access the TLSServer service."""
if self._tlsserver is None:
logger.debug("Initializing TLSServer service")
self._tlsserver = TLSServer(
xaddr=self._get_xaddr("tlsserver", "Security"), **self.common_args
)
return self._tlsserver
authorizationserver
authorizationserver() -> AuthorizationServer
Access the AuthorizationServer service.
Source code in onvif/client.py
@service
def authorizationserver(self) -> AuthorizationServer:
"""
<!-- md:version 0.0.1 -->
<!-- md:entry_point -->
Access the AuthorizationServer service."""
if self._authorizationserver is None:
logger.debug("Initializing AuthorizationServer service")
self._authorizationserver = AuthorizationServer(
xaddr=self._get_xaddr("authorizationserver", "Security"),
**self.common_args,
)
return self._authorizationserver
mediasigning
mediasigning() -> MediaSigning
Access the MediaSigning service.
Source code in onvif/client.py
@service
def mediasigning(self) -> MediaSigning:
"""
<!-- md:version 0.0.1 -->
<!-- md:entry_point -->
Access the MediaSigning service."""
if self._mediasigning is None:
logger.debug("Initializing MediaSigning service")
self._mediasigning = MediaSigning(
xaddr=self._get_xaddr("mediasigning", "Security"), **self.common_args
)
return self._mediasigning