ONVIFDiscovery
onvif.utils.ONVIFDiscovery
0.1.6 Utility class
Discover ONVIF devices using the WS-Discovery protocol.
This class provides methods for discovering ONVIF-compliant devices on the local network using WS-Discovery multicast.
Attributes:
| Name | Type | Description |
|---|---|---|
WS_DISCOVERY_PORT | int | UDP port used by WS-Discovery |
WS_DISCOVERY_ADDRESS_IPV4 | str | IPv4 multicast address used for discovery |
WS_DISCOVERY_PROBE_MESSAGE | str | SOAP probe message sent to discover devices |
NAMESPACES | dict[str, str] | XML namespaces used to parse WS-Discovery responses |
timeout | int | Discovery timeout in seconds |
interface | str | None | Network interface IP to bind to (default: auto-detect) |
Version History
- Changed in
v0.4.0:→WS_DISCOVERY_ADDRESS_IPv4WS_DISCOVERY_ADDRESS_IPV4
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
timeout | int | Discovery timeout in seconds | 4 |
interface | str | None | Network interface IP to bind to (default: auto-detect) | None |
Source code in onvif/utils/discovery.py
def __init__(
self,
timeout: int = 4,
interface: str | None = None,
):
"""Initialize ONVIF Discovery.
Args:
timeout (int): Discovery timeout in seconds
interface (str | None): Network interface IP to bind to (default: auto-detect)
"""
self.timeout = timeout
self.interface = interface
self._local_ip: str | None = None
get_local_ip
get_local_ip() -> str
Get local network interface IP address.
Returns:
| Type | Description |
|---|---|
str | Local IP address |
Source code in onvif/utils/discovery.py
def get_local_ip(self) -> str:
"""
<!-- md:version 0.1.6 -->
<!-- md:utility -->
Get local network interface IP address.
Returns:
Local IP address
"""
if self._local_ip is not None:
return self._local_ip
if self.interface:
self._local_ip = self.interface
return self._local_ip
try:
# Try to get the default route interface IP
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect(("8.8.8.8", 80))
self._local_ip = s.getsockname()[0]
s.close()
return self._local_ip
except OSError as e:
logger.debug("Failed to get local IP via default route: %s", e)
# Try alternative method to get local IP
try:
hostname = socket.gethostname()
local_ip = socket.gethostbyname(hostname)
if local_ip and not local_ip.startswith("127."):
self._local_ip = local_ip
logger.debug("Got local IP via hostname: %s", local_ip)
return self._local_ip
except OSError as exc:
logger.debug("Failed to get local IP via hostname: %s", exc)
# Return empty string instead of None for socket binding
# Empty string lets OS choose the appropriate interface
logger.debug("Using auto-detect for network interface")
self._local_ip = ""
return self._local_ip
discover
Discover ONVIF devices on the network.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
prefer_https | bool | If True, prioritize HTTPS XAddrs when available | False |
search | str | None | Optional search term to filter devices by types or scopes (case-insensitive) | None |
Returns:
| Type | Description |
|---|---|
list[dict[str, Any]] | List of discovered devices (as dict); empty if none are available |
Device dict
| Key | Type | Description |
|---|---|---|
host | str | Device IP address or hostname |
port | int | Device port number |
use_https | bool | Whether the device supports HTTPS |
epr | str | Endpoint reference |
types | list[str] | Device types |
scopes | list[str] | Device scopes |
xaddrs | list[str] | All available XAddrs |
hostname | str | None | Hostname of the device |
date_time | dict[str, str] | System date and time (utc, local); empty if none are available |
services | list[str] | List of supported ONVIF services; empty if none are available |
Version History
- Added in
v0.4.0:hostname,date_time, andservices
Source code in onvif/utils/discovery.py
def discover(
self, prefer_https: bool = False, search: str | None = None
) -> list[dict[str, Any]]:
"""
<!-- md:version 0.1.6 -->
<!-- md:utility -->
Discover ONVIF devices on the network.
Args:
prefer_https (bool): If True, prioritize HTTPS XAddrs when available
search (str | None): Optional search term to filter devices
by types or scopes (case-insensitive)
Returns:
List of discovered devices (as dict); empty if none are available
!!! abstract "Device dict"
| Key | Type | Description |
| --- | ---- | ----------- |
| `host` | `str` | Device IP address or hostname |
| `port` | `int` | Device port number |
| `use_https` | `bool` | Whether the device supports HTTPS |
| `epr` | `str` | Endpoint reference |
| `types` | `list[str]` | Device types |
| `scopes` | `list[str]` | Device scopes |
| `xaddrs` | `list[str]` | All available XAddrs |
| `hostname` | `str | None` | Hostname of the device |
| `date_time` | `dict[str, str]` | System date and time (`utc`, `local`); empty if none are available |
| `services` | `list[str]` | List of supported ONVIF services; empty if none are available |
!!! tip "Version History"
- Added in [`v0.4.0`](/onvif-python/releases/#v0.4.0): `hostname`, `date_time`, and `services`
"""
local_ip = self.get_local_ip()
logger.info("Starting ONVIF device discovery (timeout: %ss)", self.timeout)
logger.debug("Local IP: %s", local_ip or "auto-detect")
if prefer_https:
logger.debug("Prefer HTTPS endpoints enabled")
if search:
logger.debug("Search filter: %s", search)
probe_uuid = str(uuid.uuid4())
probe = self.WS_DISCOVERY_PROBE_MESSAGE.format(uuid=probe_uuid)
responses = []
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
# Bind to specific interface if available, otherwise use empty string
# Empty string lets the OS choose the appropriate interface for multicast
# This avoids the security issue of explicitly using "0.0.0.0"
bind_address = local_ip if local_ip else ""
sock.bind((bind_address, 0))
sock.settimeout(self.timeout)
ttl = struct.pack("b", 1)
sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, ttl)
logger.debug(
"Sending WS-Discovery probe to %s:%s",
self.WS_DISCOVERY_ADDRESS_IPV4,
self.WS_DISCOVERY_PORT,
)
sock.sendto(
probe.encode("utf-8"),
(self.WS_DISCOVERY_ADDRESS_IPV4, self.WS_DISCOVERY_PORT),
)
while True:
try:
data, addr = sock.recvfrom(8192)
response = data.decode("utf-8", errors="ignore").strip()
if (
response
and len(response) > 10
and response.startswith(("<?xml", "<"))
):
logger.debug("Received response from %s", addr[0])
responses.append({"xml": response, "address": addr[0]})
except socket.timeout:
logger.debug("Discovery timeout reached")
break
except (OSError, UnicodeError) as exc:
# Ignore individual packet errors and continue
logger.debug(
"Error receiving packet: %s",
exc,
)
continue
sock.close()
except OSError as exc:
# Socket creation or binding failed
logger.error("Discovery failed: %s", exc)
return []
logger.info("Received %s responses", len(responses))
# Parse responses
devices = self._parse_responses(responses, prefer_https)
# Apply search filter if provided
if search:
unfiltered_count = len(devices)
devices = self._filter_devices(devices, search)
logger.info(
"Search filter '%s' matched %s/%s devices",
search,
len(devices),
unfiltered_count,
)
# Process device information
devices = self._process_device(devices)
logger.info("Discovery completed: found %s ONVIF devices", len(devices))
return devices