Fix an issue with transport classes kwargs

Author: PaulMarisOUMary

obdii/connection.py

@@ -61,7 +61,7 @@ def __init__(
             Whether to set up the root logger.
 
         **kwargs: :class:`dict`
-            Additional keyword arguments to pass to the transport's connect method. Only used if `auto_connect` is True.
+            Additional keyword arguments forwarded to the transport's constructor.
         """
         self.transport = self._resolve_transport(transport, **kwargs)
 
@@ -131,7 +131,7 @@ def connect(self, **kwargs) -> None:
         Parameters
         ----------
         **kwargs
-            Additional parameters forwarded to the transport's `connect()` method.
+            Additional keyword arguments forwarded to the transport's connect method.
         """
         _log.info(f"Attempting to connect to {repr(self.transport)}.")
         try:

obdii/transports/transport_port.py

@@ -1,50 +1,44 @@
-from typing import Optional
+from typing import Optional, Dict, Any
 
 from serial import Serial
 
 from .transport_base import TransportBase
 
 from ..basetypes import MISSING
-from ..utils.helper import override_class_attributes
 
 
 class TransportPort(TransportBase):
-    def __init__(self, **kwargs) -> None:
-        self.overridable_attributes = {
-            "port": MISSING,
-            "baudrate": 38400,
-            "timeout": 5.0,
-            "write_timeout": 3.0,
+    def __init__(
+        self,
+        port: str = MISSING,
+        baudrate: int = 38400,
+        timeout: float = 5.0,
+        write_timeout: float = 3.0,
+        **kwargs,
+    ) -> None:
+        self.config: Dict[str, Any] = {
+            "port": port,
+            "baudrate": baudrate,
+            "timeout": timeout,
+            "write_timeout": write_timeout,
+            **kwargs,
         }
 
-        self.port: str
-        self.baudrate: int
-        self.timeout: float
-        self.write_timeout: float
-
         self.serial_conn: Optional[Serial] = None
 
-        override_class_attributes(self, self.overridable_attributes, **kwargs)
-
-        if self.port is MISSING:
+        if port is MISSING:
             raise ValueError("Port must be specified for TransportPort.")
 
     def __repr__(self) -> str:
-        return f"<TransportPort {self.port} at {self.baudrate} baud>"
+        return f"<TransportPort {self.config.get('port')} at {self.config.get('baudrate')} baud>"
 
     def is_connected(self) -> bool:
         return self.serial_conn is not None and self.serial_conn.is_open
 
     def connect(self, **kwargs) -> None:
-        override_class_attributes(self, self.overridable_attributes, True, **kwargs)
+        self.config.update(kwargs)
 
-        self.serial_conn = Serial(
-            port=self.port,
-            baudrate=self.baudrate,
-            timeout=self.timeout,
-            write_timeout=self.write_timeout,
-            **kwargs,
-        )
+        self.serial_conn = Serial(**self.config)
 
     def close(self) -> None:
         if self.serial_conn and self.serial_conn.is_open:

obdii/transports/transport_wifi.py

@@ -1,35 +1,35 @@
 from socket import AF_INET, SOCK_STREAM, error as s_error, socket
-from typing import Optional, Union
+from typing import Optional, Union, Dict, Any
 
 from .transport_base import TransportBase
 
 from ..basetypes import MISSING
-from ..utils.helper import override_class_attributes
 
 
 class TransportWifi(TransportBase):
-    def __init__(self, **kwargs) -> None:
-        self.overridable_attributes = {
-            "address": MISSING,
-            "port": MISSING,
-            "timeout": 5.0,
+    def __init__(
+        self,
+        address: str = MISSING,
+        port: Union[str, int] = MISSING,
+        timeout: float = 5.0,
+        **kwargs,
+    ) -> None:
+        self.config: Dict[str, Any] = {
+            "address": address,
+            "port": port,
+            "timeout": timeout,
+            **kwargs,
         }
 
-        self.address: str
-        self.port: Union[str, int]
-        self.timeout: float
-
         self.socket_conn: Optional[socket] = None
 
-        override_class_attributes(self, self.overridable_attributes, **kwargs)
-
-        if self.address is MISSING or self.port is MISSING:
+        if address is MISSING or port is MISSING:
             raise ValueError(
                 "Both address and port must be specified for TransportWifi."
             )
 
     def __repr__(self) -> str:
-        return f"<TransportWifi {self.address}:{self.port}>"
+        return f"<TransportWifi {self.config.get('address')}:{self.config.get('port')}>"
 
     def is_connected(self) -> bool:
         if self.socket_conn is None:
@@ -41,11 +41,15 @@ def is_connected(self) -> bool:
             return False
 
     def connect(self, **kwargs) -> None:
-        override_class_attributes(self, self.overridable_attributes, True, **kwargs)
+        self.config.update(kwargs)
+
+        timeout = self.config.get("timeout")
+        address = self.config.get("address")
+        port = self.config.get("port")
 
         self.socket_conn = socket(AF_INET, SOCK_STREAM)
-        self.socket_conn.settimeout(self.timeout)
-        self.socket_conn.connect((self.address, int(self.port)))
+        self.socket_conn.settimeout(timeout)
+        self.socket_conn.connect((address, port))
 
     def close(self) -> None:
         if self.socket_conn: