ruff formatted

Author: Lukas Kaupp

asyncua/client/client.py

@@ -23,6 +23,7 @@
 from ..common.ua_utils import value_to_datavalue, copy_dataclass_attr
 from ..crypto import uacrypto, security_policies
 from ..crypto.validator import CertificateValidatorMethod
+from ..ua.ua_binary import LowApiConfig
 
 _logger = logging.getLogger(__name__)
 
@@ -41,7 +42,7 @@ class Client:
     _password: Optional[str] = None
     strip_url_credentials: bool = True
 
-    def __init__(self, url: str, timeout: float = 4, watchdog_intervall: float = 1.0):
+    def __init__(self, url: str, timeout: float = 4, watchdog_intervall: float = 1.0, encoding="utf-8"):
         """
         :param url: url of the server.
             if you are unsure of url, write at least hostname
@@ -91,6 +92,7 @@ def __init__(self, url: str, timeout: float = 4, watchdog_intervall: float = 1.0
         self._closing: bool = False
         self.certificate_validator: Optional[CertificateValidatorMethod] = None
         """hook to validate a certificate, raises a ServiceError when not valid"""
+        LowApiConfig.String_Encoding = encoding
 
     async def __aenter__(self):
         await self.connect()

asyncua/client/ua_client.py

@@ -11,8 +11,7 @@
 from asyncua.common.session_interface import AbstractSession
 from ..common.utils import wait_for
 from asyncua.ua.uaerrors._base import UaError
-from ..ua.ua_binary import struct_from_binary, uatcp_to_binary, struct_to_binary, nodeid_from_binary, \
-    header_from_binary, LowApiConfig
+from ..ua.ua_binary import struct_from_binary, uatcp_to_binary, struct_to_binary, nodeid_from_binary, header_from_binary
 from ..ua.uaerrors import BadTimeout, BadNoSubscription, BadSessionClosed, BadUserAccessDenied, UaStructParsingError
 from ..ua.uaprotocol_auto import OpenSecureChannelResult, SubscriptionAcknowledgement
 from ..common.connection import SecureConnection, TransportLimits
@@ -289,7 +288,7 @@ class UaClient(AbstractSession):
     uaprotocol_auto.py and uaprotocol_hand.py available under asyncua.ua
     """
 
-    def __init__(self, timeout: float = 1.0, encoding = "utf-8"):
+    def __init__(self, timeout: float = 1.0):
         """
         :param timeout: Timout in seconds
         """
@@ -301,7 +300,6 @@ def __init__(self, timeout: float = 1.0, encoding = "utf-8"):
         self._publish_task = None
         self._pre_request_hook: Optional[Callable[[], Awaitable[None]]] = None
         self._closing: bool = False
-        LowApiConfig.String_Encoding = encoding
 
     def set_security(self, policy: security_policies.SecurityPolicy):
         self.security_policy = policy

asyncua/server/server.py

@@ -85,7 +85,7 @@ class Server:
         server listens on some internal IP.
     """
 
-    def __init__(self, iserver: InternalServer = None, user_manager=None, encoding = "utf-8"):
+    def __init__(self, iserver: InternalServer = None, user_manager=None, encoding="utf-8"):
         self.endpoint = urlparse("opc.tcp://0.0.0.0:4840/freeopcua/server/")
         self._application_uri = "urn:freeopcua:python:server"
         self.product_uri = "urn:freeopcua.github.io:python:server"

asyncua/ua/ua_binary.py

@@ -70,22 +70,24 @@ def unpack(data):
             return None
         return data.read(length)
 
+
 class LowApiConfig:
     String_Encoding: str = "utf-8"
 
+
 class _String:
     @staticmethod
     def pack(string):
         if string is not None:
-            string = string.encode(LowApiConfig.String_Encoding, errors='surrogateescape')
+            string = string.encode(LowApiConfig.String_Encoding, errors="surrogateescape")
         return _Bytes.pack(string)
 
     @staticmethod
     def unpack(data):
         b = _Bytes.unpack(data)
         if b is None:
             return b
-        return b.decode(LowApiConfig.String_Encoding, errors='surrogateescape')  # not need to be strict here, this is user data
+        return b.decode(LowApiConfig.String_Encoding, errors="surrogateescape")
 
 
 class _Null: