Skip to content

ONVIFDiscovery

onvif.utils.ONVIFDiscovery

ONVIFDiscovery(
    timeout: int = 4, interface: str | None = None
)

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_IPv4 → WS_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

0.1.6

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(
    prefer_https: bool = False, search: str | None = None
) -> list[dict[str, Any]]

0.1.6

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, and services
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

Comments