code(example): Simplify function docstrings and improve output formatting in device_discovery.py

Author: kaburagisec

examples/device_discovery.py

@@ -22,7 +22,6 @@
 import xml.etree.ElementTree as ET
 import sys
 import struct
-import time
 
 
 # WS-Discovery constants
@@ -48,12 +47,7 @@
 
 
 def get_network_interface():
-    """
-    Get the local network interface IP address.
-
-    Returns:
-        str: Local IP address
-    """
+    """Get the local network interface IP address."""
     try:
         # Create a socket to determine the local IP
         s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
@@ -89,10 +83,10 @@ def send_probe_and_get_responses(network_interface=None, timeout=WS_DISCOVERY_TI
     if network_interface is None:
         network_interface = get_network_interface()
 
-    print(f"Using network interface: {network_interface}")
+    print(f"Network interface: {network_interface}")
     print(f"Probe UUID: {probe_uuid}")
     print(f"Sending Probe to: {WS_DISCOVERY_ADDRESS_IPv4}:{WS_DISCOVERY_PORT}")
-    print("-" * 70)
+    print("-" * 55)
 
     responses = []
 
@@ -135,7 +129,7 @@ def send_probe_and_get_responses(network_interface=None, timeout=WS_DISCOVERY_TI
                         responses.append(
                             {"xml": response, "address": addr[0], "port": addr[1]}
                         )
-                        print(f"Received ProbeMatch from {addr[0]}:{addr[1]}")
+                        print(f"Received ProbeMatch ← {addr[0]}:{addr[1]}")
                     # else: silently ignore non-XML responses
                 # else: silently ignore empty/too short responses
 
@@ -160,15 +154,7 @@ def send_probe_and_get_responses(network_interface=None, timeout=WS_DISCOVERY_TI
 
 
 def parse_probe_match(xml_response):
-    """
-    Parse SOAP ProbeMatch XML response to extract device information.
-
-    Args:
-        xml_response (str): SOAP XML response string
-
-    Returns:
-        dict: Parsed device information or None if parsing fails
-    """
+    """Parse SOAP ProbeMatch XML response to extract device information."""
     try:
         # Clean up the XML response (remove null bytes and whitespace)
         xml_response = xml_response.strip()
@@ -255,20 +241,11 @@ def parse_probe_match(xml_response):
 
 
 def discover_onvif_devices(network_interface=None, timeout=WS_DISCOVERY_TIMEOUT):
-    """
-    Discover ONVIF devices on the network using WS-Discovery.
-
-    Args:
-        network_interface (str): Network interface IP to bind to (None for auto-detect)
-        timeout (int): Discovery timeout in seconds
-
-    Returns:
-        list: List of discovered ONVIF devices with parsed information
-    """
+    """Discover ONVIF devices on the network using WS-Discovery."""
     # Send probe and collect responses
     responses = send_probe_and_get_responses(network_interface, timeout)
 
-    print(f"\n{'='*70}")
+    print(f"\n{'-'*55}")
     print(f"Total responses received: {len(responses)}")
 
     discovered_devices = []
@@ -284,7 +261,7 @@ def discover_onvif_devices(network_interface=None, timeout=WS_DISCOVERY_TIMEOUT)
             discovered_devices.append(device_info)
 
     print(f"Valid ONVIF devices found: {len(discovered_devices)}")
-    print(f"{'='*70}\n")
+    print(f"{'-'*55}\n")
 
     # Print device information
     for device in discovered_devices:
@@ -294,31 +271,20 @@ def discover_onvif_devices(network_interface=None, timeout=WS_DISCOVERY_TIMEOUT)
 
 
 def print_device_info(device, is_onvif=True):
-    """
-    Print formatted device information from ProbeMatch response.
-
-    Args:
-        device (dict): Device information dictionary
-        is_onvif (bool): Whether this is confirmed as an ONVIF device
-    """
-    print(f"━━━ ONVIF Device #{device['index']} ━━━")
-    print(f"  Response from: {device.get('response_from', 'Unknown')}")
-    print(f"  EndpointReference (EPR):")
-    print(f"    {device['epr']}")
+    """Print formatted device information from ProbeMatch response."""
+    print(f"[#{device['index']}] - {device.get('response_from', 'Unknown')}")
+    print(f"EndpointReference (EPR):")
+    print(f"  • {device['epr']}")
 
     if device["types"]:
-        print(f"  Types (from ProbeMatch):")
+        print(f"Types (from ProbeMatch):")
         for type_info in device["types"]:
-            print(f"    • {type_info}")
-            if "NetworkVideoTransmitter" in type_info:
-                print(f"      → Camera/Video Encoder device")
-            elif "Device" in type_info:
-                print(f"      → ONVIF Device Management Service")
+            print(f"  • {type_info}")
 
     if device["xaddrs"]:
-        print(f"  Service Addresses (XAddrs):")
+        print(f"Service Addresses (XAddrs):")
         for xaddr in device["xaddrs"]:
-            print(f"    • {xaddr}")
+            print(f"  • {xaddr}")
             # Extract and display IP address and port
             if "://" in xaddr:
                 try:
@@ -329,179 +295,36 @@ def print_device_info(device, is_onvif=True):
                     if ":" in rest.split("/")[0]:
                         port_part = rest.split(":")[1].split("/")[0]
                         print(
-                            f"      → IP: {ip_part}, Port: {port_part}, Protocol: {protocol}"
+                            f"    → IP: {ip_part}, Port: {port_part}, Protocol: {protocol}"
                         )
                     else:
                         default_port = "80" if protocol == "http" else "443"
                         print(
-                            f"      → IP: {ip_part}, Port: {default_port}, Protocol: {protocol}"
+                            f"    → IP: {ip_part}, Port: {default_port}, Protocol: {protocol}"
                         )
                 except:
                     pass
 
     if device["scopes"]:
-        print(f"  Scopes (ONVIF Metadata):")
+        print(f"Scopes (ONVIF Metadata):")
         for scope in device["scopes"]:
-            print(f"    • {scope}")
-            # Parse common ONVIF scope patterns
-            if "onvif://www.onvif.org/MAC/" in scope:
-                mac = scope.split("MAC/")[1] if "MAC/" in scope else ""
-                print(f"      → MAC Address: {mac}")
-            elif "onvif://www.onvif.org/hardware/" in scope:
-                hw = scope.split("hardware/")[1] if "hardware/" in scope else ""
-                print(f"      → Hardware Model: {hw}")
-            elif "onvif://www.onvif.org/name/" in scope:
-                name = scope.split("name/")[1] if "name/" in scope else ""
-                print(f"      → Device Name: {name}")
-            elif "onvif://www.onvif.org/location/" in scope:
-                loc = scope.split("location/")[1] if "location/" in scope else ""
-                print(f"      → Location: {loc}")
-            elif "onvif://www.onvif.org/type/" in scope:
-                dtype = scope.split("type/")[1] if "type/" in scope else ""
-                print(f"      → Device Type: {dtype}")
-            elif "onvif://www.onvif.org/Profile/" in scope:
-                profile = scope.split("Profile/")[1] if "Profile/" in scope else ""
-                print(f"      → ONVIF Profile: {profile}")
+            # Remove the prefix "onvif://www.onvif.org/" if present
+            if scope.startswith("onvif://www.onvif.org/"):
+                simplified = scope.replace("onvif://www.onvif.org/", "")
+                print(f"  • [{simplified}]")
+            else:
+                # Keep other scopes as-is
+                print(f"  • [{scope}]")
 
     if device.get("metadata_version"):
-        print(f"  Metadata Version: {device['metadata_version']}")
-
-    print("-" * 70)
-
-
-def extract_device_credentials(device):
-    """
-    Extract useful connection information from discovered device.
-
-    Args:
-        device (dict): Device information dictionary
-
-    Returns:
-        dict: Connection parameters for ONVIFClient
-    """
-    connection_info = {"host": None, "port": 80, "xaddrs": device["xaddrs"]}
-
-    # Try to extract host and port from XAddrs
-    if device["xaddrs"]:
-        first_xaddr = device["xaddrs"][0]
-        try:
-            # Parse URL like http://192.168.1.100:80/onvif/device_service
-            if "://" in first_xaddr:
-                parts = first_xaddr.split("://")[1]
-                host_port = parts.split("/")[0]
-
-                if ":" in host_port:
-                    connection_info["host"] = host_port.split(":")[0]
-                    connection_info["port"] = int(host_port.split(":")[1])
-                else:
-                    connection_info["host"] = host_port
-                    connection_info["port"] = 80
-        except Exception as e:
-            print(f"Warning: Could not parse XAddr: {e}")
-
-    return connection_info
-
-
-def main():
-    """Main function to run ONVIF device discovery"""
-    print("\n" + "═" * 70)
-    print("   ONVIF Device Discovery using WS-Discovery Protocol")
-    print("   (Raw SOAP/UDP Implementation - No External Dependencies)")
-    print("═" * 70)
-    print("\nThis script sends a WS-Discovery Probe message via UDP multicast")
-    print("to discover ONVIF-compliant devices on the local network.")
-    print("\nProbe Message Details:")
-    print("  • Multicast Address: 239.255.255.250")
-    print("  • Port: 3702 (UDP)")
-    print("  • Action: http://schemas.xmlsoap.org/ws/2005/04/discovery/Probe")
-    print("  • Types: tds:Device (ONVIF Device constraint)")
-    print("  • Each probe has unique urn:uuid")
-    print()
-
-    # Discover devices (auto-detect network interface)
-    devices = discover_onvif_devices(
-        network_interface=None, timeout=WS_DISCOVERY_TIMEOUT
-    )
-
-    # Print summary
-    print("\n" + "═" * 70)
-    print(f"   Discovery Complete - Found {len(devices)} ONVIF Device(s)")
-    print("═" * 70)
-
-    if devices:
-        print("\n📋 Connection Information Summary:")
-        print("-" * 70)
-        for device in devices:
-            conn_info = extract_device_credentials(device)
-            if conn_info["host"]:
-                print(f"\n🎥 Device #{device['index']}:")
-                print(f"  Host: {conn_info['host']}")
-                print(f"  Port: {conn_info['port']}")
-
-                # Extract device info from scopes
-                device_name = "Unknown"
-                mac_address = "Unknown"
-                hardware = "Unknown"
-
-                for scope in device.get("scopes", []):
-                    if "onvif://www.onvif.org/name/" in scope:
-                        device_name = (
-                            scope.split("name/")[1] if "name/" in scope else "Unknown"
-                        )
-                    elif "onvif://www.onvif.org/MAC/" in scope:
-                        mac_address = (
-                            scope.split("MAC/")[1] if "MAC/" in scope else "Unknown"
-                        )
-                    elif "onvif://www.onvif.org/hardware/" in scope:
-                        hardware = (
-                            scope.split("hardware/")[1]
-                            if "hardware/" in scope
-                            else "Unknown"
-                        )
-
-                if device_name != "Unknown":
-                    print(f"  Name: {device_name}")
-                if hardware != "Unknown":
-                    print(f"  Model: {hardware}")
-                if mac_address != "Unknown":
-                    print(f"  MAC: {mac_address}")
-
-                print(
-                    f"  XAddr: {conn_info['xaddrs'][0] if conn_info['xaddrs'] else 'N/A'}"
-                )
-                print(f"\n  💻 Python Code to Connect:")
-                print(f"  from onvif import ONVIFClient")
-                print(
-                    f'  client = ONVIFClient("{conn_info["host"]}", {conn_info["port"]}, "admin", "password")'
-                )
-                print(f"  device = client.devicemgmt()")
-                print(f"  print(device.GetDeviceInformation())")
-        print("-" * 70)
-        print("\n💡 Tips:")
-        print("  • Replace 'admin' and 'password' with your actual credentials")
-        print("  • Most ONVIF cameras use default credentials like admin/admin")
-        print("  • Check your camera's manual for default username/password")
-    else:
-        print("\n❌ No ONVIF devices were discovered.")
-        print("\n🔍 Possible reasons:")
-        print("  • No ONVIF devices are powered on or connected to the network")
-        print("  • Devices have WS-Discovery disabled in their settings")
-        print("  • Firewall is blocking multicast UDP traffic (port 3702)")
-        print("  • Network segmentation prevents multicast discovery")
-        print("  • Devices are on a different subnet/VLAN")
-        print("\n💡 Troubleshooting:")
-        print("  • Check if devices respond to ping")
-        print("  • Verify devices are on the same network segment")
-        print("  • Try disabling firewall temporarily")
-        print("  • Check camera settings for WS-Discovery/ONVIF support")
-        print("  • Run script with administrator/root privileges")
+        print(f"Metadata Version: {device['metadata_version']}")
 
     print()
 
 
 if __name__ == "__main__":
     try:
-        main()
+        discover_onvif_devices(network_interface=None, timeout=WS_DISCOVERY_TIMEOUT)
     except KeyboardInterrupt:
         print("\n\nDiscovery interrupted by user.")
         sys.exit(0)