Use only bundled WSDL files, reduce file IO operations

Author: martyanov

onvif/cli.py

@@ -3,7 +3,6 @@
 import argparse
 import ast
 import cmd
-import os.path
 import re
 
 import zeep.exceptions
@@ -50,7 +49,6 @@ def setup(self, args):
             args.port,
             args.user,
             args.password,
-            args.wsdl,
             use_token_digest=args.encrypt,
         )
 
@@ -181,12 +179,6 @@ def create_parser():
         required=True,
         help='Password for authentication',
     )
-    parser.add_argument(
-        '-w',
-        '--wsdl',
-        default=os.path.join(os.path.dirname(os.path.dirname(__file__)), 'wsdl'),
-        help='directory to store ONVIF WSDL documents',
-    )
     parser.add_argument(
         '-e',
         '--encrypt',

onvif/client.py

@@ -1,6 +1,6 @@
 import datetime
 import logging
-import os.path
+import os
 import typing
 
 import zeep.client
@@ -103,9 +103,6 @@ def __init__(
         use_token_digest: bool = True,
         device_time_drift: typing.Optional[datetime.timedelta] = None,
     ):
-        if not os.path.isfile(url):
-            raise exceptions.ONVIFError(f"{url!r} doesn't exist!")
-
         self.url = url
         self.xaddr = xaddr
 
@@ -204,7 +201,6 @@ class ONVIFCamera:
     :param port: Camera port.
     :param username: Camera username.
     :param password: Camera user password.
-    :param wsdl_dir: Directory with WSDL definitions, use the bundled one by default.
     :param use_token_digest: Use password digest for WSSE authentication.
     :param adjust_time: Allows authentication on cameras without being time synchronized.
                         NOTE: Please note that using NTP on both end is the recommended
@@ -218,7 +214,6 @@ def __init__(
         port: int,
         username: str,
         password: str,
-        wsdl_dir: typing.Optional[str] = None,
         use_token_digest: bool = True,
         adjust_time: bool = False,
     ):
@@ -230,7 +225,6 @@ def __init__(
         self._port = port
         self._username = username
         self._password = password
-        self._wsdl_dir = wsdl_dir
         self._use_token_digest = use_token_digest
         self._adjust_time = adjust_time
 
@@ -245,10 +239,6 @@ def __init__(
 
         self.to_dict = ONVIFService.to_dict
 
-        # TODO: Better to handle with pathlib
-        if self._wsdl_dir is None:
-            self._wsdl_dir = os.path.join(os.path.dirname(__file__), 'wsdl')
-
     async def update_xaddrs(self):
         """Update xaddrs for services."""
 
@@ -306,9 +296,7 @@ def get_definition(self, name, port_type=None):
             namespace += '/' + port_type
 
         # TODO: Cache or load asynchronously
-        wsdlpath = os.path.join(self._wsdl_dir, wsdl_file)
-        if not os.path.isfile(wsdlpath):
-            raise exceptions.ONVIFError(f'No such file: {wsdlpath!r}')
+        wsdl_path = str(wsdl.WSDL_DIR / wsdl_file)
 
         # XAddr for devicemgmt is fixed
         if name == 'devicemgmt':
@@ -318,14 +306,14 @@ def get_definition(self, name, port_type=None):
                 host = f'http://{self._host}'
 
             xaddr = f'{host}:{self._port}/onvif/device_service'
-            return xaddr, wsdlpath, binding_name
+            return xaddr, wsdl_path, binding_name
 
         # Get XAddr
         xaddr = self._xaddrs.get(namespace)
         if not xaddr:
             raise exceptions.ONVIFError(f"Device doesn't support service: {name!r}")
 
-        return xaddr, wsdlpath, binding_name
+        return xaddr, wsdl_path, binding_name
 
     def create_onvif_service(
         self,

onvif/wsdl/__init__.py

@@ -1,3 +1,7 @@
+import pathlib
+
+
+WSDL_DIR = pathlib.Path(__file__).parent
 SERVICES = {
     'devicemgmt': {
         'ns': 'http://www.onvif.org/ver10/device/wsdl',